Python — Practical

Python — Practical patterns

Async with asyncio

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as r:
        return await r.text()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*(fetch(s, u) for u in urls))

asyncio.run(main(['https://a', 'https://b']))

Concurrency limit (semaphore)

sem = asyncio.Semaphore(10)
async def bounded(coro):
    async with sem:
        return await coro

Run blocking code from async

import asyncio
def blocking_sql(): ...

async def main():
    res = await asyncio.to_thread(blocking_sql)  # 3.9+

Retry with backoff (tenacity)

from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=0.2, max=5))
def call_api(): ...

Dataclasses

from dataclasses import dataclass, field
from typing import List

@dataclass(frozen=True, slots=True)
class User:
    id: str
    email: str
    roles: List[str] = field(default_factory=list)

Pydantic v2 (validation)

from pydantic import BaseModel, EmailStr, Field

class UserIn(BaseModel):
    email: EmailStr
    age: int = Field(ge=18, le=120)

u = UserIn.model_validate({'email': 'a@b.com', 'age': 25})

FastAPI minimal

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

async def get_db(): ...

@app.post('/items', status_code=201)
async def create_item(item: Item, db=Depends(get_db)):
    return await db.save(item)

Django ORM optimization

# Bad: N+1
for post in Post.objects.all():
    print(post.author.name)  # 1 query per post

# Good: FK
for post in Post.objects.select_related('author'):
    print(post.author.name)  # 1 JOIN

# Good: M2M / reverse FK
for post in Post.objects.prefetch_related('tags'):
    for t in post.tags.all(): ...

# only / defer to fetch subset of columns
Post.objects.only('id', 'title')

# values / values_list for raw dict/tuple
Post.objects.values_list('id', flat=True)

Atomic with row lock

from django.db import transaction
with transaction.atomic():
    acct = Account.objects.select_for_update().get(pk=id)
    acct.balance -= amount
    acct.save()

Logging (structured)

import logging, json
logger = logging.getLogger(__name__)

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            'level': record.levelname,
            'msg': record.getMessage(),
            'name': record.name,
        })

Context manager

from contextlib import contextmanager

@contextmanager
def db_tx(db):
    try:
        db.begin(); yield; db.commit()
    except Exception:
        db.rollback(); raise

Generators for memory efficiency

def read_lines(path):
    with open(path) as f:
        for line in f:
            yield line.rstrip()

nums = (int(l) for l in read_lines('big.txt') if l)
total = sum(nums)

Useful patterns

• functools.lru_cache(maxsize=128) — memoize pure functions.
• functools.cached_property — computed once, cached on instance.
• itertools.batched(iter, n) (3.12+) — chunked iteration.
• pathlib.Path over os.path.
• subprocess.run with check=True, capture_output=True, text=True.

Profiling commands

python -m cProfile -o prof.out app.py
snakeviz prof.out

py-spy record -o flame.svg --pid <PID>
py-spy top --pid <PID>

Useful libs

• HTTP: httpx (sync+async), aiohttp.
• DB: SQLAlchemy 2.0, asyncpg, psycopg[binary,pool].
• Queue: Celery, dramatiq, arq (async).
• Serialization: orjson, msgspec.
• Validation: pydantic, attrs.