Python dla inżynierów: głęboka semantyka 3.12‑3.14

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]

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))

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))

from itertools import pairwise

list(pairwise([1, 2, 3, 4]))
# [(1, 2), (2, 3), (3, 4)]

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:
    ...

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]

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")

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

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

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",
]

python -m cProfile -s cumulative pipeline.py