Blog › ICP guides

Python developer on retainer: asyncio architecture, type annotations, performance profiling, and packaging on monthly retainer

August 12, 2026 · ~18 min read

A fintech startup had a FastAPI service handling payment processing. The engineering team had migrated from Flask six months prior, attracted by FastAPI’s async-first design, automatic OpenAPI documentation, and Pydantic validation. The migration was syntactically correct: every route handler was declared async def, the service used FastAPI’s dependency injection system, and the Pydantic models validated all incoming request bodies. Under load, however, the service behaved like a synchronous application. Response times averaged 2.1 seconds at 50 concurrent users, CPU utilization was low, and the database connection pool was perpetually saturated — all consistent with a service that was blocking its event loop rather than yielding to it.

The team had not understood the fundamental distinction between declaring a function async def and actually writing non-blocking code inside it. Their SQLAlchemy queries used the synchronous Session and Engine — the standard SQLAlchemy 1.4 ORM — called from inside async def endpoint handlers. The synchronous session.execute(stmt) call does not yield to the event loop; it blocks the single asyncio thread for the full duration of the database round-trip. Their retry backoff logic used time.sleep(n), which blocks the OS thread entirely. Their three parallel external API calls were sequential await httpx.get(url) calls executed one at a time, each waiting for the previous to complete before starting the next.

A fractional Python architect on monthly retainer restructured the service over a single sprint. The audit phase identified every synchronous call inside an async context using PYTHONASYNCIODEBUG=1 to surface slow callbacks exceeding 100ms. The migration phase replaced create_engine with create_async_engine using the asyncpg driver, replaced every Session with AsyncSession, updated all database queries to await session.execute(stmt) inside async with session.begin() context managers, replaced every time.sleep(n) with await asyncio.sleep(n), and refactored the three sequential external API calls into a single asyncio.gather(*[fetch_rate(c) for c in codes], return_exceptions=True) with per-result exception handling. Response times dropped from 2.1 seconds average to 180 milliseconds at the same load level — a 91 percent improvement produced by fixing the event loop blocking that had been present since the Flask migration.

Python developers, Python architects, and Python consultants on monthly retainer — fractional Python engineers, Python migration consultants, and Python platform advisors — do their highest-value work in the asyncio architecture, mypy strict type annotation, performance profiling, pytest testing infrastructure, and packaging governance that produces the performant, maintainable, type-safe Python platform the engineering director reports on to the CTO. This guide covers async Python in depth, type annotations and mypy strict mode, performance profiling tools, pytest testing patterns, and modern Python packaging — and how to structure a Python developer retainer that makes the hours behind each platform function visible.

Async Python: asyncio deep dive

Python’s asyncio library provides cooperative multitasking on a single OS thread. The event loop runs one coroutine at a time; when a coroutine reaches an await expression, it suspends execution and yields control back to the event loop, which then runs another coroutine that is ready. This cooperative yielding is the mechanism that makes asyncio efficient for I/O-bound work: while one coroutine waits for a network response, another coroutine processes a completed database result. A Python architect on retainer spends significant time identifying the synchronous calls inside async contexts that break this cooperative model and block every other coroutine from running.

The event loop: coroutines, tasks, and the await chain

An async def function is a coroutine function — calling it produces a coroutine object but does not execute any of its body. The coroutine must be awaited or scheduled as a task for its body to run:

import asyncio

async def fetch_user(user_id: int) -> dict:
    # Simulates an async database call that yields to the event loop:
    await asyncio.sleep(0.05)  # Non-blocking — event loop runs other coroutines during this wait
    return {"id": user_id, "name": f"User {user_id}"}

async def main():
    # Sequential: fetch_user(1) completes fully before fetch_user(2) starts.
    # Total time: ~100ms (two 50ms waits in sequence).
    user1 = await fetch_user(1)
    user2 = await fetch_user(2)

    # Concurrent: both coroutines are scheduled as tasks and run concurrently.
    # Total time: ~50ms (both 50ms waits overlap).
    user1, user2 = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
    )

asyncio.run(main())

The critical distinction: await fetch_user(1) runs the coroutine to completion before the next line executes. asyncio.gather(fetch_user(1), fetch_user(2)) schedules both coroutines as concurrent tasks and waits for both to complete — each yields at its await asyncio.sleep(0.05), allowing the other to run. For three sequential external API calls that each take 200ms, the difference is 600ms sequential versus 200ms concurrent.

asyncio.gather with return_exceptions=True

asyncio.gather(*coros, return_exceptions=True) is the standard pattern for concurrent coroutine execution with individual exception handling. Without return_exceptions=True, the first coroutine to raise an exception cancels all remaining coroutines and raises from the gather call — partial results are lost. With it, exceptions are returned as result values in the corresponding position of the results list:

import asyncio
import httpx
from typing import Union

async def fetch_exchange_rate(client: httpx.AsyncClient, currency: str) -> dict:
    response = await client.get(f"https://api.rates.io/v1/{currency}", timeout=5.0)
    response.raise_for_status()
    return response.json()

async def fetch_all_rates(currencies: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        results: list[Union[dict, Exception]] = await asyncio.gather(
            *[fetch_exchange_rate(client, c) for c in currencies],
            return_exceptions=True,
        )

    rates = []
    for currency, result in zip(currencies, results):
        if isinstance(result, Exception):
            # Log the failure but continue with the successful results:
            print(f"Failed to fetch rate for {currency}: {result}")
        else:
            rates.append(result)
    return rates

asyncio.create_task vs asyncio.ensure_future

asyncio.create_task(coro) (Python 3.7+) is the preferred way to schedule a coroutine as a concurrent task without awaiting it immediately. Unlike await coro, create_task schedules the coroutine to run in the background and returns a Task object that can be awaited later, cancelled, or added to a set for fire-and-forget:

import asyncio

async def background_sync(record_id: int) -> None:
    await asyncio.sleep(2.0)  # Simulate a slow sync operation
    print(f"Synced record {record_id}")

async def handle_request(record_id: int) -> dict:
    # Schedule the sync as a background task — does not block the response:
    task = asyncio.create_task(background_sync(record_id))
    # Store a reference to prevent garbage collection before completion:
    background_tasks: set[asyncio.Task] = set()
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

    # Response is returned immediately; background_sync runs concurrently:
    return {"id": record_id, "status": "accepted"}

# asyncio.ensure_future is the older equivalent — accepts coroutines and futures,
# but requires an event loop to be running. Prefer create_task for coroutines.
# Task cancellation:
async def cancellable():
    task = asyncio.create_task(background_sync(99))
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled cleanly")

asyncio.Queue for producer-consumer patterns

asyncio.Queue is the async equivalent of queue.Queue — a thread-safe (or rather, coroutine-safe) bounded queue that coordinates producers and consumers without shared mutable state:

import asyncio

async def producer(queue: asyncio.Queue[int], items: list[int]) -> None:
    for item in items:
        await queue.put(item)          # Blocks if queue is full (backpressure)
    await queue.put(None)              # Sentinel value to signal completion

async def consumer(queue: asyncio.Queue[int], worker_id: int) -> None:
    while True:
        item = await queue.get()       # Suspends until an item is available
        if item is None:
            queue.task_done()
            await queue.put(None)      # Re-enqueue sentinel for other consumers
            break
        print(f"Worker {worker_id} processing {item}")
        await asyncio.sleep(0.1)       # Simulate async work
        queue.task_done()              # Signal that this item is fully processed

async def pipeline(items: list[int], num_workers: int = 3) -> None:
    # maxsize=10 provides backpressure: producer blocks when consumers fall behind
    queue: asyncio.Queue[int] = asyncio.Queue(maxsize=10)

    workers = [asyncio.create_task(consumer(queue, i)) for i in range(num_workers)]
    await producer(queue, items)

    await queue.join()   # Waits until all task_done() calls match all put() calls
    for w in workers:
        w.cancel()

asyncio.run(pipeline(list(range(50))))

asyncio.Semaphore for rate limiting

When making concurrent requests to an external API with rate limits, asyncio.Semaphore limits the number of coroutines that can be inside a critical section simultaneously — preventing 429 responses from overwhelming a concurrent fan-out:

import asyncio
import httpx

# Limit to 5 concurrent requests regardless of how many items are in the batch:
sem = asyncio.Semaphore(5)

async def fetch_with_limit(client: httpx.AsyncClient, url: str) -> dict:
    async with sem:                    # Acquires semaphore; blocks if 5 are already held
        response = await client.get(url, timeout=10.0)
        return response.json()

async def batch_fetch(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            *[fetch_with_limit(client, url) for url in urls],
            return_exceptions=True,
        )
    return [r for r in results if not isinstance(r, Exception)]

Blocking the event loop: identification and remediation

The event loop runs on a single OS thread. Any synchronous call that takes more than a few milliseconds — a database query via the standard SQLAlchemy Session, a requests.get call, a time.sleep, a CPU-intensive loop — blocks that thread and prevents every other coroutine from running. The tools for identifying blocking calls:

# Enable asyncio debug mode to print warnings for slow callbacks (>100ms):
import asyncio
import os

os.environ["PYTHONASYNCIODEBUG"] = "1"
# Or programmatically:
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.slow_callback_duration = 0.05  # Warn on callbacks >50ms

# For CPU-bound work: run in a ProcessPoolExecutor to avoid blocking the event loop:
import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_intensive(data: bytes) -> bytes:
    # Heavy computation — runs in a separate process, not the event loop thread:
    import hashlib
    return hashlib.sha256(data).digest()

async def process_upload(data: bytes) -> bytes:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_intensive, data)
    return result

# For blocking I/O in a sync library: asyncio.to_thread (Python 3.9+) is simpler:
import asyncio

def read_large_file(path: str) -> bytes:
    with open(path, "rb") as f:
        return f.read()

async def load_config(path: str) -> bytes:
    # Runs read_large_file in the default thread pool; event loop is not blocked:
    return await asyncio.to_thread(read_large_file, path)

aiohttp.ClientSession vs httpx.AsyncClient

Both libraries provide async HTTP clients, but with different design philosophies. httpx.AsyncClient is the preferred choice for new projects: it mirrors the requests API closely (reducing learning curve), supports HTTP/2, provides a synchronous httpx.Client for non-async contexts with the same API, and integrates cleanly with pytest-httpx for testing. aiohttp.ClientSession is more mature, has a larger async ecosystem, and is appropriate when deep aiohttp integration (websockets, streaming, server-sent events) is required:

import httpx
import asyncio
from typing import AsyncGenerator
from contextlib import asynccontextmanager

# httpx.AsyncClient with explicit connection pooling and timeouts:
@asynccontextmanager
async def http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    # limits: max_connections=100, max_keepalive_connections=20
    limits = httpx.Limits(max_connections=100, max_keepalive_connections=20)
    timeout = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0)
    async with httpx.AsyncClient(limits=limits, timeout=timeout) as client:
        yield client

# FastAPI lifespan event: create shared client once, reuse across requests:
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize shared resources
    app.state.http_client = httpx.AsyncClient(
        limits=httpx.Limits(max_connections=50),
        timeout=httpx.Timeout(10.0),
    )
    yield
    # Shutdown: clean up resources
    await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

FastAPI dependency injection and background tasks

from fastapi import FastAPI, Depends, BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select
from typing import Annotated

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/payments"
engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

DBSession = Annotated[AsyncSession, Depends(get_db)]

app = FastAPI()

@app.get("/payments/{payment_id}")
async def get_payment(payment_id: int, db: DBSession) -> dict:
    result = await db.execute(select(Payment).where(Payment.id == payment_id))
    payment = result.scalar_one_or_none()
    if payment is None:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Payment not found")
    return {"id": payment.id, "amount": payment.amount, "status": payment.status}

async def send_confirmation_email(payment_id: int) -> None:
    # Runs after the response is sent — does not block the client:
    await asyncio.sleep(0.1)  # Simulate async email API call
    print(f"Confirmation sent for payment {payment_id}")

@app.post("/payments/")
async def create_payment(
    payload: PaymentCreate,
    db: DBSession,
    background_tasks: BackgroundTasks,
) -> dict:
    async with db.begin():
        payment = Payment(amount=payload.amount, status="pending")
        db.add(payment)
        await db.flush()
        payment_id = payment.id
    # Schedule email after response — client receives 201 immediately:
    background_tasks.add_task(send_confirmation_email, payment_id)
    return {"id": payment_id, "status": "pending"}

SQLAlchemy 2.0 async: AsyncSession and async_sessionmaker

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Numeric
from sqlalchemy import select, update
import decimal

class Base(DeclarativeBase):
    pass

class Payment(Base):
    __tablename__ = "payments"
    id:     Mapped[int]             = mapped_column(primary_key=True)
    amount: Mapped[decimal.Decimal] = mapped_column(Numeric(12, 2))
    status: Mapped[str]             = mapped_column(String(32))

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,       # Recycle stale connections before use
    echo=False,
)

# async_sessionmaker replaces the older sessionmaker for async sessions:
Session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async def mark_payment_complete(payment_id: int) -> None:
    async with Session() as session:
        async with session.begin():   # Commits on exit, rolls back on exception
            await session.execute(
                update(Payment)
                .where(Payment.id == payment_id)
                .values(status="completed")
            )

async def get_pending_payments() -> list[Payment]:
    async with Session() as session:
        result = await session.execute(
            select(Payment).where(Payment.status == "pending").limit(100)
        )
        return list(result.scalars().all())

Python type annotations: mypy strict mode deep dive

Python’s type annotation system (PEP 484, PEP 526, PEP 544, PEP 604, and subsequent PEPs) provides optional static typing that the mypy type checker enforces. Unlike TypeScript’s mandatory type system, Python type annotations are optional at the language level — which means they accumulate only when the team actively maintains them and erode as soon as annotation discipline lapses. A Python architect on monthly retainer provides the sustained discipline that prevents annotation erosion: running mypy --strict in CI, triaging new violations introduced by PRs, and maintaining the pyproject.toml mypy configuration that defines the strictness contract.

mypy --strict configuration in pyproject.toml

# pyproject.toml
[tool.mypy]
strict = true
# strict = true enables all of the following:
# disallow_untyped_defs = true      — all functions must have type annotations
# disallow_any_generics = true      — disallow generic types without parameters (List instead of List[str])
# warn_return_any = true            — warn when a function returns Any implicitly
# disallow_subclassing_any = true   — disallow subclassing Any
# disallow_untyped_decorators = true
# warn_unused_ignores = true        — error on unnecessary # type: ignore comments
# warn_redundant_casts = true
# no_implicit_reexport = true       — __init__.py must explicitly re-export names

# Per-module overrides for third-party libraries without stubs:
[[tool.mypy.overrides]]
module = ["boto3.*", "botocore.*"]
ignore_missing_imports = true

[[tool.mypy.overrides]]
module = ["legacy_internal_sdk.*"]
ignore_missing_imports = true
disallow_untyped_defs = false  # Relax for vendored code you don't control

TypeVar with bound and Protocol for structural subtyping

from typing import TypeVar, Protocol, runtime_checkable

# TypeVar with bound: T must be a subclass of Comparable.
# Enables writing generic functions that work on any type with comparison operators:
class Comparable(Protocol):
    def __lt__(self, other: object) -> bool: ...
    def __le__(self, other: object) -> bool: ...

T = TypeVar("T", bound=Comparable)

def min_value(a: T, b: T) -> T:
    return a if a < b else b

result = min_value(3, 7)        # T inferred as int
result = min_value("a", "z")    # T inferred as str
# min_value(3, "z")             # mypy error: incompatible types

# Protocol for structural subtyping — no inheritance required:
@runtime_checkable
class Serializable(Protocol):
    def serialize(self) -> bytes: ...

class PaymentRecord:
    def serialize(self) -> bytes:
        return b"payment data"

class AuditLog:
    def serialize(self) -> bytes:
        return b"audit data"

def write_to_storage(obj: Serializable) -> None:
    data = obj.serialize()
    # ... write data

write_to_storage(PaymentRecord())  # Works — PaymentRecord satisfies Serializable
write_to_storage(AuditLog())       # Works — AuditLog satisfies Serializable
# No inheritance from Serializable required; structural compatibility is sufficient.

@dataclass with field, frozen=True, and __post_init__

from dataclasses import dataclass, field
from typing import ClassVar
import decimal

@dataclass(frozen=True)   # Immutable: all fields are read-only after __init__
class Money:
    amount: decimal.Decimal
    currency: str
    # ClassVar fields are not included in __init__ or __repr__:
    SUPPORTED_CURRENCIES: ClassVar[frozenset[str]] = frozenset(["USD", "EUR", "GBP"])

    def __post_init__(self) -> None:
        # Validation runs after __init__; raises ValueError for invalid state:
        if self.currency not in self.SUPPORTED_CURRENCIES:
            raise ValueError(f"Unsupported currency: {self.currency!r}")
        if self.amount < decimal.Decimal("0"):
            raise ValueError("Amount must be non-negative")
        # frozen=True prevents direct assignment; use object.__setattr__ for derived fields:
        object.__setattr__(self, "amount", self.amount.quantize(decimal.Decimal("0.01")))

    def __add__(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Cannot add amounts in different currencies")
        return Money(self.amount + other.amount, self.currency)

@dataclass
class PaymentBatch:
    payments: list[Money] = field(default_factory=list)   # Mutable default via factory
    metadata: dict[str, str] = field(default_factory=dict, repr=False)  # Excluded from repr
    _total: decimal.Decimal = field(default=decimal.Decimal("0"), init=False, repr=False)

    def add(self, payment: Money) -> None:
        self.payments.append(payment)
        self._total += payment.amount

TypedDict, Annotated, and Pydantic v2 validators

from typing import TypedDict, Annotated
from typing import Required, NotRequired  # Python 3.11+; use typing_extensions on 3.9/3.10

# TypedDict for structured dictionaries with per-key optionality:
class UserBase(TypedDict):
    id: int
    email: str
    name: str

class UserUpdate(UserBase, total=False):
    # total=False: all keys are optional in UserUpdate
    # Inheriting from UserBase with total=False allows mixing required and optional:
    role: str
    avatar_url: str

# Inline mixing of required and optional (Python 3.11+):
class PaymentRequest(TypedDict):
    amount: Required[int]
    currency: Required[str]
    description: NotRequired[str]   # Optional — may be absent

# Annotated[T, metadata] for Pydantic v2 field validators:
from pydantic import BaseModel, Field, field_validator, model_validator

class CreatePaymentRequest(BaseModel):
    amount: Annotated[int, Field(gt=0, le=1_000_000, description="Amount in cents")]
    currency: Annotated[str, Field(min_length=3, max_length=3, pattern="^[A-Z]{3}$")]
    description: Annotated[str, Field(default="", max_length=500)]
    idempotency_key: Annotated[str, Field(min_length=1, max_length=255)]

    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        supported = {"USD", "EUR", "GBP", "JPY"}
        if v not in supported:
            raise ValueError(f"Currency must be one of {supported}")
        return v

    @model_validator(mode="after")
    def validate_jpy_whole_units(self) -> "CreatePaymentRequest":
        # JPY has no decimal subdivision — amount must be in whole units:
        if self.currency == "JPY" and self.amount % 100 != 0:
            raise ValueError("JPY amounts must be whole units (multiples of 100 cents)")
        return self

ParamSpec, Literal, Final, and overload

from typing import ParamSpec, TypeVar, Callable, Literal, Final, overload
import functools
import logging

P = ParamSpec("P")
T = TypeVar("T")

# ParamSpec preserves the parameter types of the wrapped function through the decorator:
def log_calls(fn: Callable[P, T]) -> Callable[P, T]:
    @functools.wraps(fn)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        logging.info(f"Calling {fn.__name__}")
        result = fn(*args, **kwargs)
        logging.info(f"{fn.__name__} returned")
        return result
    return wrapper

@log_calls
def process_payment(amount: int, currency: str) -> dict:
    return {"amount": amount, "currency": currency}

# mypy knows process_payment still accepts (amount: int, currency: str) -> dict:
process_payment(100, "USD")    # OK
# process_payment("bad", 100) # mypy error: argument type mismatch

# Literal for string enum alternatives:
PaymentStatus = Literal["pending", "completed", "failed", "refunded"]

def update_status(payment_id: int, status: PaymentStatus) -> None:
    # mypy rejects any string not in the Literal:
    pass

update_status(1, "completed")   # OK
# update_status(1, "cancelled") # mypy error: not a valid PaymentStatus

# Final for module-level constants:
MAX_RETRY_ATTEMPTS: Final[int] = 3
API_VERSION: Final[str] = "v2"

# overload for multiple function signatures:
@overload
def get_payment(payment_id: int) -> dict: ...
@overload
def get_payment(payment_id: str) -> dict | None: ...

def get_payment(payment_id: int | str) -> dict | None:
    if isinstance(payment_id, int):
        return {"id": payment_id}
    return None  # String IDs may not exist

Python performance: profiling and optimization

Performance work in Python follows a strict methodology: measure first, optimize second. The Python architect on retainer does not guess at hotspots — they profile the actual production workload with cProfile, line_profiler, memory_profiler, or py-spy before writing a single optimization. Premature optimization in Python is particularly expensive: the language’s dynamism means that micro-optimizations that seem obvious (loop unrolling, manual caching) often produce negligible improvement while numpy vectorization or a strategic lru_cache produces a 10x speedup that profiling reveals.

cProfile and line_profiler

# cProfile: function-level profiling. Run from command line:
# python -m cProfile -s cumulative -o profile.out my_script.py
# python -m pstats profile.out

# Or programmatically:
import cProfile
import pstats
import io

def profile_function(fn, *args, **kwargs):
    pr = cProfile.Profile()
    pr.enable()
    result = fn(*args, **kwargs)
    pr.disable()

    s = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
    ps.print_stats(20)   # Top 20 functions by cumulative time
    print(s.getvalue())
    return result

# line_profiler: line-level profiling for identified hotspot functions.
# Install: pip install line_profiler
# Usage: kernprof -l -v my_script.py
# Decorate the function to profile:

# @profile  — added by kernprof at runtime, remove before committing
def compute_risk_score(transactions: list[dict]) -> float:
    total = 0.0
    for txn in transactions:                 # Line 1: hot
        amount = txn["amount"]               # Line 2: hot — dict access in loop
        category = txn.get("category", "")  # Line 3: hot
        if category == "high_risk":
            total += amount * 2.5
        else:
            total += amount
    return total / len(transactions) if transactions else 0.0

memory_profiler, __slots__, and generator expressions

# memory_profiler: line-level memory usage.
# Install: pip install memory_profiler
# mprof run my_script.py  — records memory over time
# mprof plot              — generates a timeline plot

# @profile decorator (added by memory_profiler, not runtime Python):
# from memory_profiler import profile
# @profile
def load_transactions(filepath: str) -> list[dict]:
    # List comprehension: builds entire list in memory at once
    return [parse_line(line) for line in open(filepath)]

# Generator expression: processes one item at a time — constant memory usage:
def stream_transactions(filepath: str):
    return (parse_line(line) for line in open(filepath))

# __slots__ reduces per-instance memory by replacing the instance __dict__
# with a fixed-size C-level slot array. Critical for classes with millions of instances:

class TransactionSlow:
    # Without __slots__: each instance has a __dict__ (~200-400 bytes overhead)
    def __init__(self, id: int, amount: float, currency: str) -> None:
        self.id = id
        self.amount = amount
        self.currency = currency

class TransactionFast:
    __slots__ = ("id", "amount", "currency")  # Eliminates __dict__; ~3x less memory per instance
    def __init__(self, id: int, amount: float, currency: str) -> None:
        self.id = id
        self.amount = amount
        self.currency = currency

# Memory impact at scale (1 million instances):
# TransactionSlow: ~350 bytes/instance → ~350 MB
# TransactionFast: ~120 bytes/instance → ~120 MB
# Savings: ~230 MB for 1M instances

functools.lru_cache, functools.cache, and invalidation

import functools
from typing import Optional

# lru_cache(maxsize=N): bounded LRU cache; evicts least-recently-used entries
@functools.lru_cache(maxsize=128)
def get_exchange_rate(from_currency: str, to_currency: str) -> float:
    # Expensive: calls an external API. Cached by (from_currency, to_currency) pair.
    import time
    time.sleep(0.1)  # Simulate network call
    return 1.08  # EUR/USD placeholder

# functools.cache (Python 3.9+): unbounded cache — equivalent to lru_cache(maxsize=None)
# Use only when key space is provably finite (e.g., Enum members, small fixed sets):
@functools.cache
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Cache inspection and manual invalidation:
get_exchange_rate.cache_info()     # CacheInfo(hits=5, misses=2, maxsize=128, currsize=2)
get_exchange_rate.cache_clear()    # Invalidate the entire cache

# Thread safety: lru_cache IS thread-safe for reads (the cache lookup is atomic).
# The decorated function may be called multiple times concurrently for the same key
# before the first call completes (cache stampede). For async code, use aiocache or
# a custom asyncio.Lock-based memoizer:

import asyncio
from typing import Any

_cache: dict[tuple, Any] = {}
_locks: dict[tuple, asyncio.Lock] = {}

async def cached_async(key: tuple, coro_fn, *args) -> Any:
    if key not in _locks:
        _locks[key] = asyncio.Lock()
    async with _locks[key]:
        if key not in _cache:
            _cache[key] = await coro_fn(*args)
    return _cache[key]

GIL, multiprocessing, and concurrent.futures

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
from typing import Callable, TypeVar
import asyncio

# GIL implications:
# - CPU-bound work: GIL prevents true parallelism in threads. Use ProcessPoolExecutor.
# - I/O-bound blocking work (legacy sync library): GIL is released during I/O. Use ThreadPoolExecutor.
# - I/O-bound async work: Use asyncio directly (single thread, no pool needed).

def cpu_bound_chunk(data: list[int]) -> int:
    """Runs in a subprocess — not affected by GIL."""
    return sum(x ** 2 for x in data)

def parallel_cpu_work(dataset: list[int], num_chunks: int = 4) -> int:
    chunk_size = len(dataset) // num_chunks
    chunks = [dataset[i:i+chunk_size] for i in range(0, len(dataset), chunk_size)]

    with ProcessPoolExecutor(max_workers=num_chunks) as pool:
        futures = [pool.submit(cpu_bound_chunk, chunk) for chunk in chunks]
        return sum(f.result() for f in as_completed(futures))

# asyncio integration: run ProcessPoolExecutor from async context:
async def async_cpu_work(dataset: list[int]) -> int:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = await asyncio.gather(
            *[loop.run_in_executor(pool, cpu_bound_chunk, chunk)
              for chunk in [dataset[:500], dataset[500:]]]
        )
    return sum(results)

# ThreadPoolExecutor for blocking I/O (e.g., legacy sync database drivers):
def sync_db_query(query: str) -> list[dict]:
    import time
    time.sleep(0.05)  # Simulates blocking DB call
    return [{"result": query}]

async def concurrent_sync_queries(queries: list[str]) -> list[list[dict]]:
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor(max_workers=10) as pool:
        results = await asyncio.gather(
            *[loop.run_in_executor(pool, sync_db_query, q) for q in queries]
        )
    return list(results)

numpy vectorization and production profiling with py-spy

import numpy as np

# Python loop: slow — interpreter overhead on every iteration
def compute_returns_slow(prices: list[float]) -> list[float]:
    returns = []
    for i in range(1, len(prices)):
        returns.append((prices[i] - prices[i-1]) / prices[i-1])
    return returns

# numpy vectorization: single C-level operation across entire array
def compute_returns_fast(prices: np.ndarray) -> np.ndarray:
    return np.diff(prices) / prices[:-1]

# For 1 million prices: ~800ms Python loop vs ~4ms numpy — 200x speedup.
# The key: avoid Python-level iteration on large numeric datasets entirely.

# py-spy: sampling profiler for production Python processes (no code changes needed):
# py-spy top --pid 12345                  — live top-like view of hot functions
# py-spy record --pid 12345 -o trace.svg  — generates a flame graph SVG
# py-spy dump --pid 12345                 — thread stack dump without restart

# pyinstrument: call-graph profiler for development; lower overhead than cProfile:
# pip install pyinstrument
# python -m pyinstrument my_script.py
# Or programmatically:
from pyinstrument import Profiler

def profile_section():
    with Profiler() as profiler:
        # ... code to profile
        pass
    profiler.print()  # Prints a call tree sorted by wall time

pytest: fixtures, parametrize, hypothesis, and async testing

A Python architect on retainer establishing testing infrastructure for a FastAPI service designs the pytest fixture hierarchy, async testing configuration, and property-based testing strategy that enables the engineering team to maintain coverage discipline as the service grows. The investment is front-loaded — 8 to 20 hours to establish the conftest.py shared fixtures, async session factories, and hypothesis strategy libraries — and pays dividends in the reduced per-test setup code that follows.

Fixtures: scope, composition, and conftest.py

# tests/conftest.py — shared fixtures available to all tests in the project:
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.pool import NullPool

TEST_DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/test_db"

@pytest.fixture(scope="session")
def event_loop_policy():
    """Use uvloop event loop policy for tests (matches production)."""
    import uvloop
    return uvloop.EventLoopPolicy()

@pytest_asyncio.fixture(scope="session")
async def db_engine():
    """Session-scoped engine: created once for the entire test session."""
    engine = create_async_engine(
        TEST_DATABASE_URL,
        poolclass=NullPool,   # NullPool: no connection reuse between tests
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest_asyncio.fixture(scope="function")
async def db_session(db_engine) -> AsyncSession:
    """Function-scoped session: each test gets a fresh, rolled-back transaction."""
    async with db_engine.connect() as conn:
        await conn.begin_nested()          # Savepoint for rollback
        session = AsyncSession(bind=conn)
        yield session
        await session.close()
        await conn.rollback()              # Roll back all changes after each test

pytest.mark.parametrize for data-driven tests

import pytest
from decimal import Decimal

@pytest.mark.parametrize("amount,currency,expected_error", [
    (100,   "USD", None),          # Valid
    (0,     "USD", "gt=0"),        # Zero amount: fails gt=0 validator
    (-50,   "EUR", "gt=0"),        # Negative: fails gt=0
    (100,   "XX",  "pattern"),     # Invalid currency: fails pattern
    (100,   "usd", "pattern"),     # Lowercase: fails pattern ^[A-Z]{3}$
    (1_500_000, "USD", "le=1000000"),  # Exceeds max
])
def test_payment_validation(amount, currency, expected_error):
    from pydantic import ValidationError
    from app.schemas import CreatePaymentRequest

    if expected_error is None:
        req = CreatePaymentRequest(amount=amount, currency=currency, idempotency_key="k1")
        assert req.amount == amount
    else:
        with pytest.raises(ValidationError) as exc_info:
            CreatePaymentRequest(amount=amount, currency=currency, idempotency_key="k1")
        assert expected_error in str(exc_info.value)

# Parametrize with IDs for readable test output:
@pytest.mark.parametrize("status,is_terminal", [
    pytest.param("pending",   False, id="pending-not-terminal"),
    pytest.param("completed", True,  id="completed-is-terminal"),
    pytest.param("failed",    True,  id="failed-is-terminal"),
    pytest.param("refunded",  True,  id="refunded-is-terminal"),
])
def test_payment_status_terminality(status: str, is_terminal: bool) -> None:
    from app.domain import is_terminal_status
    assert is_terminal_status(status) == is_terminal

Hypothesis for property-based testing

from hypothesis import given, settings, HealthCheck
from hypothesis import strategies as st
from decimal import Decimal

# Property-based test: for any valid (amount, currency) pair,
# Money addition must be commutative and associative:
@given(
    a=st.integers(min_value=1, max_value=1_000_000),
    b=st.integers(min_value=1, max_value=1_000_000),
    currency=st.sampled_from(["USD", "EUR", "GBP"]),
)
@settings(max_examples=500)
def test_money_addition_commutative(a: int, b: int, currency: str) -> None:
    from app.domain import Money
    m_a = Money(Decimal(a), currency)
    m_b = Money(Decimal(b), currency)
    assert m_a + m_b == m_b + m_a   # Commutativity: a + b == b + a

@given(
    amounts=st.lists(st.integers(min_value=1, max_value=10_000), min_size=2, max_size=20),
    currency=st.sampled_from(["USD", "EUR"]),
)
@settings(max_examples=200, suppress_health_check=[HealthCheck.too_slow])
def test_batch_total_equals_sum_of_parts(amounts: list[int], currency: str) -> None:
    from app.domain import Money, PaymentBatch
    batch = PaymentBatch()
    for amount in amounts:
        batch.add(Money(Decimal(amount), currency))
    expected = sum(amounts)
    assert batch._total == Decimal(expected)

# Hypothesis automatically finds edge cases: empty strings, max integers,
# unicode boundaries, NaN, signed zeros — cases a human test author misses.
# @settings(max_examples=500) increases the search budget beyond the default 100.

AsyncMock, pytest-asyncio, and pytest-cov

# pyproject.toml: configure pytest-asyncio to auto-detect async tests:
# [tool.pytest.ini_options]
# asyncio_mode = "auto"         — all async test functions marked automatically
# addopts = "--cov=src --cov-report=term-missing --cov-fail-under=80"

import pytest
from unittest.mock import patch, AsyncMock, MagicMock

@pytest.mark.asyncio
async def test_payment_service_calls_external_api() -> None:
    from app.services.payment import PaymentService

    # AsyncMock: automatically awaitable, works with async with / async for:
    mock_http = AsyncMock()
    mock_http.post.return_value.__aenter__.return_value.json = AsyncMock(
        return_value={"transaction_id": "txn_abc123", "status": "approved"}
    )

    with patch("app.services.payment.httpx.AsyncClient", return_value=mock_http):
        service = PaymentService()
        result = await service.charge(amount=1000, currency="USD")

    assert result["transaction_id"] == "txn_abc123"
    mock_http.post.assert_called_once()

# patch.object for patching a specific method on a class:
@pytest.mark.asyncio
async def test_payment_retries_on_timeout() -> None:
    from app.services.payment import PaymentService
    import httpx

    service = PaymentService()
    call_count = 0

    async def flaky_post(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise httpx.TimeoutException("timeout")
        return MagicMock(json=lambda: {"status": "approved"})

    with patch.object(service._client, "post", side_effect=flaky_post):
        result = await service.charge(amount=500, currency="USD")

    assert call_count == 3   # Two failures, one success
    assert result["status"] == "approved"

# capfd for capturing stdout/stderr in tests:
def test_logging_output(capfd) -> None:
    from app.services.audit import log_event
    log_event("payment_created", {"id": 1})
    out, err = capfd.readouterr()
    assert "payment_created" in out

# tmp_path for temporary filesystem:
def test_csv_export(tmp_path) -> None:
    from app.export import export_payments_csv
    output_file = tmp_path / "payments.csv"
    export_payments_csv([{"id": 1, "amount": 100}], output_file)
    assert output_file.exists()
    assert "amount" in output_file.read_text()

Python packaging and toolchain governance

Python packaging is a retainer function that is almost entirely invisible to stakeholders. When it works, dependencies resolve reproducibly, CI runs deterministically, and new team members can set up their development environment in minutes. When it breaks — dependency conflicts, non-reproducible builds, slow pip install times in CI, setup.py legacy configuration that is incompatible with the current build backend — it consumes engineering hours disproportionate to the apparent simplicity of the problem. A Python architect on monthly retainer governs the packaging toolchain so it never becomes a blocking issue.

pyproject.toml with [project] metadata and optional dependencies

# pyproject.toml — single source of truth for project metadata, dependencies, and tool config:
[project]
name = "payment-service"
version = "0.5.2"
description = "Async payment processing service"
requires-python = ">=3.11"
dependencies = [
    "fastapi>=0.111.0",
    "sqlalchemy[asyncio]>=2.0.30",
    "asyncpg>=0.29.0",
    "httpx>=0.27.0",
    "pydantic>=2.7.0",
    "uvicorn[standard]>=0.29.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.2.0",
    "pytest-asyncio>=0.23.0",
    "pytest-cov>=5.0.0",
    "hypothesis>=6.100.0",
    "mypy>=1.10.0",
    "ruff>=0.4.0",
]
profiling = [
    "memory-profiler>=0.61.0",
    "line-profiler>=4.1.0",
    "pyinstrument>=4.6.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/payment_service"]

# ruff: replaces flake8 + isort + black (10-100x faster):
[tool.ruff]
line-length = 100
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "ANN"]
ignore = ["ANN101"]  # Missing type annotation for self

uv for fast dependency resolution

# uv: a Rust-based package installer and resolver — 10-100x faster than pip.
# Key commands:

# Create a virtual environment (replaces python -m venv):
uv venv

# Install dependencies from pyproject.toml (replaces pip install -e .):
uv pip install -e ".[dev,profiling]"

# Generate a lockfile from pyproject.toml dependency specifications:
uv lock
# Produces uv.lock — commit this to version control for reproducible installs.

# Install from lockfile (CI: fast, reproducible, no resolution needed):
uv sync

# Add a dependency and update the lockfile atomically:
uv add httpx
uv add --dev pytest-httpx

# Run a command in the project's virtual environment:
uv run pytest
uv run mypy src/

# Speed comparison (cold cache, 50-package project):
# pip install: 45-90 seconds
# uv sync:     2-5 seconds (with lockfile resolution skipped)
# The 10-100x speedup comes from uv's parallel download, Rust-native SAT solver,
# and global package cache that avoids re-downloading across projects.

nox for multi-environment testing and CI

# noxfile.py — defines test sessions across Python versions and tools:
import nox

PYTHON_VERSIONS = ["3.11", "3.12"]

@nox.session(python=PYTHON_VERSIONS)
def tests(session: nox.Session) -> None:
    """Run pytest with coverage across supported Python versions."""
    session.install(".[dev]")
    session.run(
        "pytest",
        "--cov=src",
        "--cov-report=term-missing",
        "--cov-fail-under=80",
        "-x",   # Stop on first failure
    )

@nox.session(python="3.12")
def typecheck(session: nox.Session) -> None:
    """Run mypy --strict type checking."""
    session.install(".[dev]")
    session.run("mypy", "--strict", "src/")

@nox.session(python="3.12")
def lint(session: nox.Session) -> None:
    """Run ruff linter and formatter check."""
    session.install("ruff")
    session.run("ruff", "check", "src/")
    session.run("ruff", "format", "--check", "src/")

# Run all sessions:
# nox
# Run specific sessions:
# nox -s tests-3.11 tests-3.12
# nox -s typecheck

src layout, importlib.resources, and namespace packages

# src layout prevents accidental imports of uninstalled package code:
# payment-service/
#   src/
#     payment_service/
#       __init__.py
#       api/
#       domain/
#       services/
#       data/           # Package data files
#         templates/
#           email.html
#   tests/
#   pyproject.toml

# importlib.resources for accessing package data files (Python 3.9+):
from importlib.resources import files

def load_email_template() -> str:
    # Works correctly regardless of whether the package is installed
    # as a directory, a zip, or a wheel — unlike __file__-relative paths:
    template_path = files("payment_service.data.templates").joinpath("email.html")
    return template_path.read_text(encoding="utf-8")

# Implicit namespace packages (PEP 420): packages without __init__.py.
# Useful for splitting a namespace (e.g., company.payments, company.auth)
# across multiple distributions. Each sub-package is a separate pip install
# but they share the same namespace prefix:
# company-payments/src/company/payments/__init__.py  (no company/__init__.py)
# company-auth/src/company/auth/__init__.py          (no company/__init__.py)
# Both install as sub-packages of the `company` namespace.

Python retainer work patterns: what gets underlogged

The challenge with Python retainer work is that its most valuable outputs are architecture changes, type annotation migrations, and profiling investigations that produce no deployable feature. A Python architect who spends 12 hours auditing async endpoint handlers, migrating SQLAlchemy to the 2.0 async API, and replacing sequential API calls with asyncio.gather has produced an 11x latency improvement — but the retainer invoice line shows 12 hours of “asyncio migration” against zero visible new features. The engineering director reviewing the invoice needs to connect those hours to the p95 latency dropping from 2.1 seconds to 180 milliseconds.

The async architecture audit that preceded the event loop fix

Before a single line of async code was changed in the fintech startup’s FastAPI service, the Python architect spent three hours on an audit phase: enabling PYTHONASYNCIODEBUG=1 and setting loop.slow_callback_duration = 0.05 to surface every callback exceeding 50 milliseconds, writing a locust load test that produced realistic concurrent request patterns, running the load test against the staging environment with asyncio debug logging enabled, and systematically mapping which endpoint handlers were blocking the event loop and for how long. That audit phase produced a ranked list of blocking calls by total time cost — the input to the migration plan. It also confirmed that the issue was event loop blocking rather than database query performance, infrastructure capacity, or network latency. The audit produced no artifact visible to the CTO beyond the subsequent migration PR. Without a work log entry describing the three audit hours — what was tested, what was found, how the diagnosis was confirmed — those hours are invisible in the performance improvement metric.

The mypy strict migration that preceded the type error elimination

Running mypy --strict on a Python codebase that has never had strict = true in its configuration typically produces 200 to 600 violations, depending on codebase size and annotation discipline. The Python architect’s migration work follows a structured path: first, add disallow_untyped_defs = true alone and fix only the missing return type annotations and parameter annotations — typically the largest category of violations; second, add warn_return_any = true and trace the Any return types back to their source (often missing stubs for third-party libraries); third, add disallow_any_generics = true and replace bare List, Dict, and Optional with parameterized equivalents (list[str], dict[str, int], str | None). Each phase is a separate PR with a focused fix scope. The full migration for a medium-sized FastAPI service typically takes 12 to 25 hours spread across two or three retainer weeks — invisible in the pyproject.toml one-line strict = true addition that represents the end state.

The memory_profiler session that preceded the __slots__ optimization

When a payment processing service’s memory usage grew from 180 MB to 650 MB over three months with no apparent change in traffic, the Python architect ran a memory_profiler investigation. The mprof run timeline showed memory growing steadily in proportion to the number of TransactionRecord instances held in the in-memory event buffer — a class without __slots__ with 12 instance attributes, producing a per-instance memory footprint three times larger than necessary. The profiling session took four hours: two hours instrumenting the service, running representative traffic against the staging environment, and capturing the mprof plot timeline; one hour tracing the memory growth to the specific class using tracemalloc; one hour implementing __slots__ on TransactionRecord and its two parent classes, verifying that no code relied on dynamic attribute assignment that __slots__ prevents. The resulting PR was three lines of code. The profiling investigation that made those three lines the right three lines took four hours and produced no artifact beyond the commit.

HourTab for Python developer retainers

Python developer retainer work produces FastAPI services with 180ms p95 latency instead of 2.1 second averages, mypy strict codebases where every function signature is annotated and Any propagation is blocked at the CI gate, memory usage profiles where __slots__ and generator expression substitutions reduce per-instance overhead by 60 percent, and pytest suites with 80 percent coverage thresholds enforced in CI. The hours behind each outcome are invisible without a work log that connects them: the asyncio architecture audit that preceded the event loop fix, the mypy migration that preceded the type error elimination, the memory_profiler session that preceded the __slots__ optimization.

HourTab gives Python architects and Python consultants a retainer dashboard their engineering directors can bookmark without creating an account: the month’s committed hours, the hours consumed, and the work log entries that connect each block to the specific Python platform function performed. When the VP Engineering can see that 12 of the month’s 35 retainer hours went to the asyncio migration of the payment service endpoint handlers, 8 went to mypy strict mode violation triage and resolution across the domain layer, and 4 went to the memory profiling investigation that identified the TransactionRecord.__slots__ optimization, the retainer renewal conversation is grounded in the actual distribution of Python platform advisory work rather than an abstract sense of whether the architecture investment produced value.

The retainer model fits Python architecture consulting because Python codebases are living systems: every new module added by a developer without async discipline reintroduces synchronous blocking calls inside async handlers; every dependency upgrade in the SQLAlchemy, Pydantic, or FastAPI ecosystem brings API changes that require migration; every Python version release (historically annual) introduces new type annotation syntax (X | Y union syntax in 3.10, asyncio.to_thread in 3.9, tomllib in 3.11, typing.override in 3.12) that the Python architect evaluates for adoption; and the mypy strict configuration requires ongoing maintenance as new stubs packages are released and per-module overrides are resolved. A monthly hour commitment provides the Python architect’s sustained availability across the full async platform maintenance and evolution calendar.

For Python consultants documenting retainer work, sharing a live hours dashboard replaces the weekly status email: the client sees the current month’s hour consumption and the work log entries that narrate what each block of Python advisory hours accomplished — and why the 12 hours of asyncio migration that produced a 91 percent latency improvement were worth every minute of the retainer.

Frequently asked questions

What does a Python developer on retainer typically do?

A Python developer or Python architect on monthly retainer provides ongoing Python platform advisory and development across four principal service areas. Asyncio architecture and FastAPI optimization: auditing async endpoint handlers for synchronous blocking calls that stall the event loop (sync SQLAlchemy queries, time.sleep, blocking filesystem I/O inside async def functions), migrating to SQLAlchemy 2.0 async engine with AsyncSession and async with session.begin(), replacing blocking patterns with asyncio.to_thread or loop.run_in_executor for CPU-bound work, and designing asyncio.gather and asyncio.Semaphore patterns for concurrent external API calls with rate limiting. mypy strict type annotation and Pydantic v2 migration: running mypy --strict across the codebase, fixing disallow_untyped_defs and warn_return_any violations, introducing TypeVar with bound constraints, Protocol for structural subtyping, ParamSpec for decorator type preservation, and Annotated[T, Field(...)] validators for Pydantic v2 model fields. Performance profiling and optimization: identifying hotspots with cProfile and line_profiler, reducing memory usage with __slots__ and generator expressions, implementing functools.lru_cache and functools.cache for computation memoization, using concurrent.futures.ProcessPoolExecutor for CPU-bound parallel work, and profiling production services with py-spy. Testing infrastructure and packaging: establishing pytest fixtures with conftest.py, pytest.mark.parametrize for data-driven tests, hypothesis for property-based testing, AsyncMock and pytest-asyncio for async test coverage, and pyproject.toml governance with uv or hatch for reproducible dependency resolution.

What Python work is most commonly underlogged in a retainer?

The most systematically underlogged categories are asyncio architecture audits (identifying synchronous blocking calls inside async endpoint handlers, mapping which database queries, HTTP calls, and filesystem operations stall the event loop, designing the corrective migration path — produces no visible artifact beyond the subsequent performance improvement; typically 8 to 18 hours of investigation and planning before a single line of async code is changed); mypy strict migration (running mypy --strict, triaging the resulting errors by module criticality, resolving disallow_untyped_defs violations that require rethinking function signatures, adding Protocol definitions for structural interfaces — typically 12 to 25 hours per project invisible in the pyproject.toml one-line strict = true addition); memory profiling sessions (running memory_profiler across service hot paths, identifying the specific class instances consuming disproportionate memory, designing the __slots__ migration and generator expression substitutions — typically 6 to 14 hours per service invisible in the memory reduction metric); and pytest infrastructure establishment (writing conftest.py shared fixtures, adding AsyncMock layers for external service dependencies, configuring pytest-asyncio with asyncio_mode = auto, setting up pytest-cov with coverage thresholds — typically 8 to 20 hours per project invisible in the coverage percentage reported in CI). Detailed work log entries that capture the specific blocking calls identified, mypy violations resolved, and memory hotspots profiled connect the invisible Python platform investment to its concrete outcomes.

What should a Python developer retainer agreement include?

Python developer retainer agreements should specify: scope boundary between feature development, architecture advisory, code review, and migration work (asyncio refactoring and mypy strict migration produce no deployable feature — define these explicitly as in-scope functions with their own hour allocation); repository and environment access level required (read access for audit and code review; write access for pull request authorship; CI pipeline access for mypy and pytest configuration; production metrics access for py-spy sampling during profiling sessions); toolchain scope (whether the retainer covers pyproject.toml governance, uv or hatch migration, tox or nox multi-environment configuration, and ruff linting setup); IP ownership for utility libraries, Protocol definitions, typed dataclass hierarchies, and hypothesis test strategies authored during the engagement; asyncio governance scope (who owns event loop configuration, lifespan event management, connection pool sizing for the async database engine, and Semaphore concurrency limits for external API rate limiting); and a shared work log documenting each asyncio audit session, mypy migration milestone, profiling investigation, and pytest infrastructure sprint. Monthly retainer amounts for Python developer advisory and architecture consulting typically range from $4,500 to $9,000 per month for code review and advisory retainers, increasing to $10,000 to $22,000 per month for full-stack Python architecture consulting covering async migration, strict type annotation, performance profiling, and packaging governance.

What are typical retainer rates for Python developers and Python architects?

Entry-level Python developers with 1 to 3 years of experience, Python proficiency, and FastAPI or Django integration skill typically bill $70 to $125 per hour, with monthly retainers running 10 to 18 hours for code review and advisory work. Mid-level Python engineers with 3 to 8 years of experience, expertise in async Python (asyncio, aiohttp, SQLAlchemy 2.0 async), type annotations with mypy strict, and pytest testing infrastructure typically bill $115 to $200 per hour, with monthly retainers running 15 to 30 hours. Senior Python architects with 8 to 14 years of experience, expertise in async architecture, performance profiling with cProfile and py-spy, Python distribution and packaging governance, and CPython internals including GIL implications and memory model, typically bill $175 to $340 per hour, with monthly retainers running 20 to 40 hours. Python consulting firms and specialized backend architecture consultancies typically bill $145 to $260 per hour. Monthly retainer amounts range from $4,500 to $9,000 per month for code review and advisory retainers, increasing to $10,000 to $22,000 per month for full-stack Python architecture consulting engagements covering async migration, mypy strict type annotation, performance profiling, and packaging governance.

How should Python developer retainer hours be logged?

Work log entries should capture the advisory category (asyncio architecture, mypy type annotation, performance profiling, pytest infrastructure, packaging governance, code review), the specific service or module, the task, and the finding or deliverable. Example: “Asyncio Architecture — payment-service, app/api/routers/payments.py. Task: audit async endpoint handlers for synchronous event loop blocking; migrate SQLAlchemy calls to async engine. Work: audited 8 async endpoint handlers using PYTHONASYNCIODEBUG=1 to surface slow callbacks; found sync SQLAlchemy Session.execute() calls inside every async def handler blocking the event loop for 80–400ms per query — 3 hours; replaced create_engine with create_async_engine using asyncpg driver; updated all queries to await session.execute(stmt) inside async with session.begin() — 4 hours; replaced three time.sleep(n) retry backoff calls with await asyncio.sleep(n) — 1 hour; refactored three sequential external API calls to asyncio.gather with return_exceptions=True — 3 hours; load-tested before and after with locust; p95 latency dropped from 2.1s to 180ms at 50 concurrent users — 1 hour. Total: 12 hours. Runtime behavior: unchanged. p95 latency improvement: 91%. Event loop blocking eliminated: 100% of identified synchronous database calls.” Entries that document the specific blocking calls identified and the latency improvement measured connect the 12 hours of async architecture work to the performance outcome it produced, making the Python retainer investment legible to the engineering director reviewing the work log.