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.
- Mutowalne argumenty domyślne: dlaczego ewaluują się raz i jak tego uniknąć wzorcem None-sentinel.
- GIL i free-threaded build (PEP 703/779): kiedy threading, multiprocessing, a kiedy asyncio.
- Generatory vs listy: leniwa ewaluacja i pamięć O(1) kontra O(n).
- Dekoratory sparametryzowane i functools.wraps dla poprawnej introspekcji.
- match/case: capture pattern, wzorce klas/sekwencji/mapowań, guard i OR-pattern.
- Modelowanie danych (dataclass/NamedTuple/Pydantic) i exception groups (except*).
1 · 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 kazdym wywolaniu # bez walrusa: dwa odczyty tej samej wartosci
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
2 · 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 moze nie istniec / zawsze True
if hasattr(sys, "_is_gil_enabled"):
print(sys._is_gil_enabled()) # False w binarce python3.13t / 3.14t bez ext. wymuszajacych 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 obowiazkowy: worker (spawn, domyslny na macOS/Windows) re-importuje ten modul
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ę |
3 · 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 polaczenie kilku iterowalnych w jeden strumien
- 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)] 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__)
5 · 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 tlumimy wyjatku, 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 zamkniete automatycznie przy wyjsciu, w tym przy wyjatku
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 wyjatki
- contextlib.redirect_stdout/redirect_stderr - tymczasowe przekierowanie strumieni
- contextlib.nullcontext() - placeholder gdy menedzer kontekstu jest opcjonalny
6 · 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") - Uzywaj match/case gdy masz kilka strukturalnie roznych ksztaltow danych do rozroznienia (np. rozne typy zdarzen)
- Dla prostego porownania kilku wartosci skalarnych zwykle if/elif jest rownie czytelne i nie wymaga uczenia sie skladni wzorcow
- Klasa musi definiowac __match_args__ (dataclass generuje to automatycznie) aby wspierac pozycyjne wzorce klas jak Point(0, 0)
7 · 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 - dziala tez jak zwykla 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
- Czesc stdlib, zero zewnetrznych zaleznosci
- Zero narzutu walidacji w runtime - zle dane wchodza bez ostrzezenia
- Optymalne gdy dane pochodza z zaufanego zrodla (wewnetrzna logika)
Pydantic BaseModel
- Wymaga zaleznosci (pydantic), rdzen walidacji napisany w Rust (pydantic-core) dla wydajnosci
- Waliduje i konwertuje typy w runtime, rzuca ValidationError przy zlych danych
- Optymalne na granicy systemu: API request/response, parsowanie configu, dane z zewnatrz
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 znaja typ, ale runtime nie sprawdza niczego | Narzedzie | Twórca / status | Domyslna surowosc | Typowe uzycie |
|---|---|---|---|
| mypy | Referencyjny checker PEP 484, community | umiarkowana, konfigurowalna stopniowo | duze bazy kodu z bogatym ekosystemem pluginow (Django, SQLAlchemy) |
| pyright | Microsoft, uzywany w VS Code (Pylance) | wyzsza domyslnie, szybszy silnik | szybka petla feedbacku w edytorze, CI |
8 · Nowoczesna składnia, pakowanie i wydajność
★ egzaminOd Python 3.11 do 3.14 seria mniejszych, ale wysoko wykorzystywanych w praktyce zmian: bogatsza gramatyka f-stringow, grupy wyjatkow do obslugi wspolbieznych awarii, deklaratywne pakowanie przez pyproject.toml i tansze narzedzia do profilowania.
name = "world"
# ten sam znak cudzyslowu wewnatrz wyrazenia f-stringa - blad przed 3.12
greeting = f"hello {"mr. " + name}"
# wyrazenie wieloliniowe z komentarzem wewnatrz {}
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)} bledne 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 calego wywolania, python -m cProfile -s cumulative script.py
- timeit (stdlib) - mikrobenchmarki krotkich fragmentow kodu, izoluje szum systemowy
- py-spy - profiler probkujacy dolaczany do dzialajacego procesu bez zmian w kodzie, bezpieczny na produkcji
- sys.monitoring (PEP 669, od 3.12) - nowe, tansze API zdarzen dla profilerow/debuggerow/narzedzi coverage, zastepuje kosztowny sys.settrace
python -m cProfile -s cumulative pipeline.py Sprawdź się - testowanie to nauka
24 pytań w losowej kolejności. Twoje wyniki zapisują się lokalnie.