Python Cheatsheet: Comprehensions, Typing and Async Reference

# Manage Python itself
pyenv install 3.13.1            # if you use pyenv
uv python install 3.13          # if you use uv (recommended)

# Per-project virtualenv
python -m venv .venv && source .venv/bin/activate     # stdlib
uv venv && source .venv/bin/activate                  # uv (faster)

# Dependencies
pip install -r requirements.txt                       # classic
uv pip install -r requirements.txt                    # uv (drop-in, ~10x)
uv add httpx pydantic                                 # add + lock + install

# Run
python -m my_pkg.cli                                  # treat as module
python -c "import sys; print(sys.version_info)"

from dataclasses import dataclass, field, asdict, replace
from typing import Self

@dataclass(slots=True, frozen=True, kw_only=True)
class User:
    id: int
    name: str
    email: str
    tags: list[str] = field(default_factory=list)

    @classmethod
    def from_row(cls, row: dict) -> Self:
        return cls(id=row["id"], name=row["name"], email=row["email"])

u = User(id=1, name="Ada", email="ada@example.com")
asdict(u)                       # {'id': 1, 'name': 'Ada', ...}
u2 = replace(u, name="Ada L.")  # frozen-safe copy with overrides
hash(u)                         # works because frozen=True

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> int:
    r = await client.get(url, timeout=5.0)
    return r.status_code

async def main(urls: list[str]) -> list[int]:
    async with httpx.AsyncClient() as client:
        # asyncio.TaskGroup (3.11+) — structured concurrency
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(client, u)) for u in urls]
        return [t.result() for t in tasks]

# Run from a sync entry point
results = asyncio.run(main(["https://example.com", "https://example.org"]))

# Bound concurrency with a Semaphore
sem = asyncio.Semaphore(8)

async def bounded(client, url):
    async with sem:
        return await fetch(client, url)

from pathlib import Path

p = Path("data/inputs/log.txt")
p.parent.mkdir(parents=True, exist_ok=True)

p.write_text("hello\n", encoding="utf-8")
content = p.read_text(encoding="utf-8")

# Glob / iterate
for f in Path("data").rglob("*.json"):
    print(f.relative_to("data"))

# Compose paths
out = Path.home() / "exports" / f"{p.stem}.csv"

# Inspect
p.exists(), p.is_file(), p.suffix, p.stem, p.name

# Streaming line read
with p.open(encoding="utf-8") as fh:
    for line in fh:
        ...

from contextlib import contextmanager, asynccontextmanager, ExitStack
import time

@contextmanager
def timed(label: str):
    t0 = time.perf_counter()
    try:
        yield
    finally:
        print(f"{label}: {time.perf_counter() - t0:.3f}s")

with timed("query"):
    result = run_query()

# Open many files; close them all even if one open() raises
with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]
    ...

# Async variant — perfect for httpx / DB sessions
@asynccontextmanager
async def session():
    s = await open_session()
    try:
        yield s
    finally:
        await s.close()

# Fetch JSON from N endpoints in parallel, validate, write a CSV summary.
import asyncio, csv
from pathlib import Path
import httpx
from pydantic import BaseModel, HttpUrl, ValidationError

class Repo(BaseModel):
    full_name: str
    stargazers_count: int
    html_url: HttpUrl

async def fetch_repo(client: httpx.AsyncClient, name: str) -> Repo | None:
    r = await client.get(f"https://api.github.com/repos/{name}", timeout=10)
    if r.status_code != 200:
        return None
    try:
        return Repo.model_validate(r.json())
    except ValidationError:
        return None

async def main(repos: list[str], out: Path) -> None:
    async with httpx.AsyncClient(headers={"Accept": "application/vnd.github+json"}) as c:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_repo(c, r)) for r in repos]
    rows = [t.result() for t in tasks if t.result()]
    out.parent.mkdir(parents=True, exist_ok=True)
    with out.open("w", newline="", encoding="utf-8") as fh:
        w = csv.writer(fh)
        w.writerow(["repo", "stars", "url"])
        for r in rows:
            w.writerow([r.full_name, r.stargazers_count, str(r.html_url)])

if __name__ == "__main__":
    asyncio.run(main(
        ["python/cpython", "django/django", "pallets/flask"],
        Path("out/repos.csv"),
    ))