Wszystkie kursy / Programowanie
Python dla inżynierów: głęboka semantyka 3.12‑3.14
Zaawansowana, aktualna semantyka Pythona 3.12‑3.14: GIL i współbieżność, generatory, dekoratory, dopasowanie wzorców, modelowanie danych, typowanie statyczne i wydajność dla praktykujących inżynierów.
Tematy egzaminacyjne w tym kursie
- 01 Mutowalne argumenty domyślne: dlaczego ewaluują się raz i jak tego uniknąć wzorcem None-sentinel.
- 02 GIL i free-threaded build (PEP 703/779): kiedy threading, multiprocessing, a kiedy asyncio.
- 03 Generatory vs listy: leniwa ewaluacja i pamięć O(1) kontra O(n).
- 04 Dekoratory sparametryzowane i functools.wraps dla poprawnej introspekcji.
- 05 match/case: capture pattern, wzorce klas/sekwencji/mapowań, guard i OR-pattern.
- 06 Modelowanie danych (dataclass/NamedTuple/Pydantic) i exception groups (except*).
Domyślne argumenty i wyrażenia przypisania
★ EgzaminArgument domyślny w Pythonie jest ewaluowany dokładnie raz, w momencie wykonania instrukcji def, a nie przy każdym wywołaniu funkcji. Rezultat tej jednorazowej ewaluacji trafia do atrybutu __defaults__ obiektu funkcji i jest reużywany przy każdym wywołaniu bez podanego argumentu. Dla typów mutowalnych (list, dict, set) to prosta droga do buga produkcyjnego: różne wywołania dzielą ten sam obiekt.
def add_item(item, bucket=[]):
bucket.append(item)
return bucket
add_item("a") # ['a']
add_item("b") # ['a', 'b'] - ta sama lista, nie nowa!
def add_item_fixed(item, bucket=None):
if bucket is None:
bucket = []
bucket.append(item)
return bucket
add_item_fixed("a") # ['a']
add_item_fixed("b") # ['b'] - nowa lista przy każdym wywołaniu # bez walrusa: dwa odczyty tej samej wartości
chunk = file.read(8192)
while chunk:
process(chunk)
chunk = file.read(8192)
# z walrusem: jeden odczyt, czytelniejszy warunek
while (chunk := file.read(8192)):
process(chunk)
# w liscie skladanej: policz raz, uzyj dwa razy
results = [y for x in data if (y := expensive(x)) is not None] - Używaj walrusa gdy unikasz podwójnej ewaluacji tego samego wyrażenia (np. odczyt, regex match)
- Nie nadużywaj go do upychania wielu przypisań w jednej linii kosztem czytelności
- W warunkach while i if wewnątrz comprehension jest to idiomatyczne, w zwykłym kodzie liniowym zwykłe przypisanie bywa czytelniejsze
GIL, wątki, procesy i asyncio
★ EgzaminCPython (referencyjna implementacja) chroni swoje wewnętrzne struktury danych jednym globalnym zamkiem: GIL (Global Interpreter Lock). W danym momencie tylko jeden wątek może wykonywać bytecode Pythona, niezależnie od liczby rdzeni CPU. To nie jest szczegół implementacyjny bez znaczenia, to fundamentalnie kształtuje wybór modelu współbieżności.
import sys
# w standardowym buildzie atrybut może nie istnieć / zawsze True
if hasattr(sys, "_is_gil_enabled"):
print(sys._is_gil_enabled()) # False w binarce python3.13t / 3.14t bez rozszerzeń wymuszających GIL from concurrent.futures import ThreadPoolExecutor
import httpx
def fetch_size(url: str) -> int:
return len(httpx.get(url).content)
urls = ["https://example.com"] * 20
with ThreadPoolExecutor(max_workers=20) as pool:
sizes = list(pool.map(fetch_size, urls)) from concurrent.futures import ProcessPoolExecutor
def is_prime(n: int) -> bool:
return n > 1 and all(n % i for i in range(2, int(n ** 0.5) + 1))
# guard obowiązkowy: worker (spawn, domyślny na macOS/Windows) re-importuje ten moduł
if __name__ == "__main__":
with ProcessPoolExecutor() as pool:
results = list(pool.map(is_prime, range(10_000_000, 10_000_200))) import asyncio
import httpx
async def fetch_size(client: httpx.AsyncClient, url: str) -> int:
response = await client.get(url)
return len(response.content)
async def main(urls: list[str]) -> list[int]:
async with httpx.AsyncClient() as client:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(fetch_size(client, u)) for u in urls]
return [t.result() for t in tasks]
asyncio.run(main(["https://example.com"] * 1000)) | Model | Najlepszy dla | Jednostka izolacji | Typowy narzut | Kluczowe ograniczenie |
|---|---|---|---|---|
| threading | I/O-bound (sieć, dysk, DB) | wątek OS, współdzielona pamięć | niski | GIL serializuje bytecode - brak realnego przyspieszenia CPU-bound |
| multiprocessing | CPU-bound (obliczenia, ML preprocessing) | proces OS, osobna pamięć | wysoki (start procesu, pickling, IPC) | brak współdzielonego stanu, komunikacja przez kolejki/pickle |
| asyncio | masowa współbieżność I/O (tysiące połączeń) | coroutine w jednej pętli zdarzeń | bardzo niski | wymaga bibliotek async-native, blokujące wywołanie zawiesza całą pętlę |
Generatory i iteratory: pamięć kontra listy
★ EgzaminLista buduje i przechowuje wszystkie elementy naraz w pamięci, zużycie rośnie liniowo z rozmiarem danych (O(n)). Generator produkuje elementy leniwie, jeden na żądanie, przez protokół iteratora, zużycie pamięci pozostaje stałe (O(1)) niezależnie od rozmiaru źródła danych.
def read_large_log(path: str):
with open(path) as f:
for line in f:
yield line.rstrip("\n")
errors = sum(1 for line in read_large_log("app.log") if "ERROR" in line) # ~0.8 GB tymczasowej listy zanim sum zacznie liczyc
total = sum([x * x for x in range(10**8)])
# stala, minimalna pamiec - jeden element generatora na raz
total = sum(x * x for x in range(10**8)) - itertools.islice(gen, n) - leniwe pobranie n elementów bez materializacji reszty
- itertools.chain(a, b) - leniwe połączenie kilku iterowalnych w jeden strumień
- itertools.groupby(data, key) - grupowanie kolejnych elementow (wymaga posortowanych danych wg klucza)
- itertools.pairwise(data) - kolejne pary sasiadujacych elementow (od Python 3.10)
from itertools import pairwise
list(pairwise([1, 2, 3, 4]))
# [(1, 2), (2, 3), (3, 4)] Dekoratory i functools
★ EgzaminDekorator to funkcja wyższego rzędu przyjmująca funkcję i zwracająca (zwykle) inną funkcję ją opakowującą. Składnia @decorator nad definicją funkcji to czysty cukier składniowy dla func = decorator(func), wykonywany raz, w momencie definicji.
def timed(func):
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
print(f"{func.__name__} took {time.perf_counter() - start:.4f}s")
return result
return wrapper
@timed
def compute(n: int) -> int:
"""Sum of squares up to n."""
return sum(i * i for i in range(n))
compute.__name__ # 'wrapper' - zle, powinno byc 'compute'
compute.__doc__ # None - docstring zgubiony import functools
import time
def timed(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
print(f"{func.__name__} took {time.perf_counter() - start:.4f}s")
return result
return wrapper
@timed
def compute(n: int) -> int:
"""Sum of squares up to n."""
return sum(i * i for i in range(n))
compute.__name__ # 'compute'
compute.__doc__ # 'Sum of squares up to n.' import functools
import time
def retry(times: int = 3, delay: float = 0.5):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exc = None
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except ConnectionError as exc:
last_exc = exc
time.sleep(delay)
raise last_exc
return wrapper
return decorator
@retry(times=5, delay=1.0)
def call_flaky_api() -> dict:
... - @staticmethod / @classmethod - metody nie/związane z instancją
- @functools.cached_property - property liczone raz i zapamiętane na instancji
- @functools.singledispatch - dekorator generyczny dispatchujący po typie pierwszego argumentu
- @functools.total_ordering - dogenerowuje pozostale operatory porownania z jednego (__eq__ + __lt__)
Menedżery kontekstu i contextlib
Instrukcja with gwarantuje wywołanie sprzątania niezależnie od tego, czy blok zakończył się normalnie, czy wyjątkiem. Kontrakt jest prosty: __enter__ zwraca wartość dla as, __exit__ dostaje informację o wyjątku (albo None) i decyduje, czy go wygasić.
class Transaction:
def __init__(self, conn):
self.conn = conn
def __enter__(self):
self.conn.execute("BEGIN")
return self.conn
def __exit__(self, exc_type, exc_value, traceback):
if exc_type is None:
self.conn.execute("COMMIT")
else:
self.conn.execute("ROLLBACK")
return False # nie tłumimy wyjątku, propagujemy dalej
with Transaction(conn) as cur:
cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1") from contextlib import contextmanager
@contextmanager
def transaction(conn):
conn.execute("BEGIN")
try:
yield conn
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
with transaction(conn) as cur:
cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1") from contextlib import ExitStack
def process_files(paths: list[str]):
with ExitStack() as stack:
files = [stack.enter_context(open(p)) for p in paths]
# wszystkie pliki zamknięte automatycznie przy wyjściu, w tym przy wyjątku
return [f.read() for f in files] - contextlib.closing(obj) - wywoluje obj.close() na wyjsciu dla obiektow bez pelnego protokolu with
- contextlib.suppress(*exc_types) - ignoruje wskazane wyjątki
- contextlib.redirect_stdout/redirect_stderr - tymczasowe przekierowanie strumieni
- contextlib.nullcontext() - placeholder gdy menedzer kontekstu jest opcjonalny
Dopasowanie wzorców: match/case
★ Egzaminmatch/case (PEP 634-636, od Python 3.10) to strukturalne dopasowanie wzorców, nie prosty switch po wartości. Potrafi jednocześnie sprawdzić kształt danych (typ, długość, obecność kluczy) i wydobyć z nich wartości do zmiennych w jednym kroku.
def http_status_label(code: int) -> str:
match code:
case 200 | 201 | 204:
return "OK"
case 400 | 404 | 422:
return "Client error"
case 500:
return "Server error"
case _:
return "Unknown" from dataclasses import dataclass
@dataclass
class Point:
x: int
y: int
def describe(shape) -> str:
match shape:
case Point(x=0, y=0):
return "origin"
case Point(x=0, y=y):
return f"on y-axis at {y}"
case Point(x=x, y=y) if x == y:
return "on diagonal"
case Point():
return "generic point"
case _:
return "not a point" def handle(event: dict | list) -> None:
match event:
case {"type": "click", "x": x, "y": y}:
print(f"click at ({x}, {y})")
case {"type": "key", "code": code}:
print(f"key {code}")
case [first, *rest] if rest:
print(f"batch starting with {first}, {len(rest)} more")
case []:
print("empty batch") - Używaj match/case, gdy masz kilka strukturalnie różnych kształtów danych do rozróżnienia (np. różne typy zdarzeń)
- Dla prostego porównania kilku wartości skalarnych zwykłe if/elif jest równie czytelne i nie wymaga uczenia się składni wzorców
- Klasa musi definiowac __match_args__ (dataclass generuje to automatycznie) aby wspierac pozycyjne wzorce klas jak Point(0, 0)
Modelowanie danych i typowanie statyczne
★ EgzaminWybór między @dataclass, typing.NamedTuple a Pydantic BaseModel to wybór między trzema różnymi kompromisami: zero zależności i brak walidacji (dataclass), niemutowalność i lekkość krotki (NamedTuple), albo walidacja w runtime kosztem zależności zewnętrznej (Pydantic).
from dataclasses import dataclass, field
@dataclass(slots=True, frozen=True)
class Config:
name: str
tags: list[str] = field(default_factory=list)
retries: int = 3 from typing import NamedTuple
class Point(NamedTuple):
x: int
y: int
p = Point(1, 2)
p.x # 1
p[0] # 1 - działa też jak zwykła krotka from pydantic import BaseModel, Field
class UserCreate(BaseModel):
email: str
age: int = Field(ge=0, le=150)
UserCreate(email="a@b.com", age=200)
# ValidationError: age - Input should be less than or equal to 150 dataclass / NamedTuple
- Część biblioteki standardowej, zero zewnętrznych zależności
- Zero narzutu walidacji w runtime - złe dane wchodzą bez ostrzeżenia
- Optymalne, gdy dane pochodzą z zaufanego źródła (wewnętrzna logika)
Pydantic BaseModel
- Wymaga zależności (pydantic), rdzeń walidacji napisany w Rust (pydantic-core) dla wydajności
- Waliduje i konwertuje typy w runtime, rzuca ValidationError przy złych danych
- Optymalne na granicy systemu: API request/response, parsowanie configu, dane z zewnątrz
from typing import Protocol, runtime_checkable
@runtime_checkable
class Renderable(Protocol):
def render(self) -> str: ...
class Button:
def render(self) -> str:
return "<button/>"
def show(item: Renderable) -> None:
print(item.render())
show(Button()) # OK - Button pasuje strukturalnie, nie dziedziczy po Renderable from typing import TypedDict
class UserPayload(TypedDict):
id: int
email: str
is_admin: bool
def handle(payload: UserPayload) -> None:
print(payload["email"]) # pyright/mypy znają typ, ale runtime nie sprawdza niczego | Narzędzie | Twórca / status | Domyślna surowość | Typowe użycie |
|---|---|---|---|
| mypy | Referencyjny checker PEP 484, community | umiarkowana, konfigurowalna stopniowo | duże bazy kodu z bogatym ekosystemem wtyczek (Django, SQLAlchemy) |
| pyright | Microsoft, używany w VS Code (Pylance) | wyższa domyślnie, szybszy silnik | szybka pętla informacji zwrotnej w edytorze, CI |
Nowoczesna składnia, pakowanie i wydajność
★ EgzaminOd Python 3.11 do 3.14 seria mniejszych, ale wysoko wykorzystywanych w praktyce zmian: bogatsza gramatyka f-stringów, grupy wyjątków do obsługi współbieżnych awarii, deklaratywne pakowanie przez pyproject.toml i tańsze narzędzia do profilowania.
name = "world"
# ten sam znak cudzysłowu wewnątrz wyrażenia f-stringa - błąd przed 3.12
greeting = f"hello {"mr. " + name}"
# wyrażenie wieloliniowe z komentarzem wewnątrz {}
value = f"{
name.upper() # normalizuj przed wstawieniem
}" import asyncio
async def fetch(url: str) -> str:
if "bad" in url:
raise ValueError(f"invalid url: {url}")
return url
async def main():
try:
async with asyncio.TaskGroup() as tg:
for url in ["ok1", "bad1", "ok2", "bad2"]:
tg.create_task(fetch(url))
except* ValueError as eg:
print(f"{len(eg.exceptions)} błędne URL-e")
asyncio.run(main()) [build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "tenzan-inspector"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"httpx>=0.27",
"pydantic>=2.7",
] - cProfile (stdlib) - profilowanie deterministyczne całego wywołania, python -m cProfile -s cumulative script.py
- timeit (stdlib) - mikrobenchmarki krótkich fragmentów kodu, izoluje szum systemowy
- py-spy - profiler próbkujący dołączany do działającego procesu bez zmian w kodzie, bezpieczny na produkcji
- sys.monitoring (PEP 669, od 3.12) - nowe, tańsze API zdarzeń dla profilerów, debuggerów i narzędzi coverage, zastępuje kosztowny sys.settrace
python -m cProfile -s cumulative pipeline.py Fiszki
aktywne przypominanie
Sprawdź się - testowanie to nauka
20 pytań losowanych z 24, w losowej kolejności. Twoje wyniki zapisują się na tym urządzeniu.