DEV Community: Huy Nguyen

Benchmarking Python API Frameworks with Real Workloads: FastAPI vs Litestar vs DRF vs Django Ninja vs Django Bolt

Huy Nguyen — Tue, 11 Aug 2026 13:42:01 +0000

Most Python framework benchmarks measure "hello world" JSON. Real APIs do authentication,
database access and complex queries - so that's what I benchmarked, across FastAPI,
Litestar, Django REST Framework, Django Ninja and Django Bolt, each with multiple
production servers, under strict Docker resource limits.

Everything is reproducible: https://github.com/huynguyengl99/python-api-frameworks-benchmark

The core design decisions

Latest versions of everything - Django Bolt 0.10, Django 6.0, FastAPI 0.141, Litestar 2.24, DRF 3.18, Django Ninja 1.6.2.
Each framework uses its own native serializer. Feeding Litestar Pydantic models makes it convert twice (Pydantic then msgspec), so you'd be benchmarking Pydantic instead of Litestar - a point the Litestar author raised on an earlier version of this benchmark. Litestar and Bolt use msgspec structs, FastAPI and Ninja use Pydantic, DRF uses its serializers. The response models are defined twice, field for field, and the output is byte-identical: 4,796 bytes for /articles/1 from both FastAPI and Litestar.
JWT in an httpOnly cookie for the auth endpoints - the setup I'd actually use in production. Each framework uses its own ecosystem's library instead of something hand-rolled: drf-auth-kit (DRF), django-ninja-jwt (Ninja), AuthX (FastAPI), and the built-in cookie JWT support in Litestar and Bolt.
Median across container restarts, not repeated runs. Some servers (notably Bolt's Rust worker pool under a 1-CPU cgroup) pick a throughput regime at container startup and hold it, so best-of-N inside one container flatters the lucky starts. Details in the repo's Methodology section.

Setup

MacBook M2 Pro, 32GB RAM; PostgreSQL 16; 500 articles, 2000 comments, 100 tags, 50 authors
Each framework alone in a Docker container: --cpus=1, --memory=750m, one at a time
bombardier, 100 connections, 10s per measurement
5 independent container starts per framework, median reported with min-max spread

Endpoints:

Endpoint	What it measures
`/json-1k`, `/json-10k`	Pure serialization
`/db`	10 simple rows
`/articles?page=1&page_size=20`	Pagination + nested author/tags
`/articles/1`	Nested author + tags + comments
`/auth/me`	Cookie JWT verify + load user
`/auth/articles`	Cookie JWT + the paginated query

One fairness detail that matters: all five frameworks load the user row from the DB on
both auth endpoints. Litestar, DRF and Ninja do it as part of authenticating; AuthX and
Bolt's guards only verify the signature, so I made FastAPI and Bolt load the user
explicitly. Otherwise they'd do strictly less work and the comparison would be junk.

Results

RPS, median over 5 container starts, zero errors across all 70 measurements:

Config	json-1k	json-10k	/db	/articles	/articles/1	/auth/me	/auth/articles
bolt	38,576	19,089	1,986	208	432	3,024	196
litestar-uvicorn	31,284	24,547	1,039	246	443	976	193
litestar-granian	19,006	15,166	1,180	250	488	1,104	210
fastapi-uvicorn	13,845	2,641	984	224	428	820	193
fastapi-granian	8,484	2,280	952	201	410	749	199
drf-gunicorn	3,925	3,132	282	140	193	261	133
drf-granian	2,703	2,200	830	198	321	726	179
ninja-granian	1,566	1,422	680	130	295	610	117
ninja-uvicorn	1,533	1,424	699	126	236	584	114
drf-uvicorn	1,035	973	495	153	234	447	137

The gap collapses once you hit the database

37x between fastest and slowest on /json-1k. 1.9x on /articles/1. If your endpoint
touches PostgreSQL, your framework is not the bottleneck - your queries are.

What authentication actually costs

Comparing /auth/articles against the identical unauthenticated /articles:

Config	public	authed	cost
fastapi-granian	201	199	-1%
drf-gunicorn	140	133	-5%
bolt	208	196	-6%
drf-granian	198	179	-10%
ninja-uvicorn	126	114	-10%
fastapi-uvicorn	224	193	-14%
litestar-granian	250	210	-16%
litestar-uvicorn	246	193	-22%

Roughly 5-20%, cheaper than I expected for "verify a token and load a user". Two things
stand out:

Bolt's /auth/me is 2.7x ahead of everything (3,024 vs 1,104) because it validates the JWT in Rust before Python runs at all.
Litestar pays the most for a concrete reason: its JWTCookieAuth runs in middleware, before dependency injection, so retrieve_user_handler opens its own DB session - that request pays for two connection acquisitions instead of one.

Memory and CPU

Most configs peak at 195-260MB RAM
drf-granian is the outlier at 456MB - untuned --blocking-threads, per the Granian maintainer (see "unaddressed feedback" below)
Nearly everything saturates ~85% of the 1-CPU budget under load - except Bolt at 67%, while leading most endpoints

A word on Django Bolt

If you're open to a young framework, this is the one I'd watch. It won or tied the top
spot on 4 of 7 endpoints while leaving ~18% more CPU headroom than every other config -
room to grow under load, not just a good number on a chart. And unlike moving to Litestar
or FastAPI, you keep the Django ORM, admin and package ecosystem.

Honest trade-offs: it's young and moving fast, the Rust internals (Actix Web + PyO3 +
msgspec) mean you can't monkey-patch your way out of a corner, and under a hard 1-CPU cap
its throughput varies between container starts under a hard 1-CPU cap. For a side project or a new
internal service, I'd have no problem reaching for it today.

Feedback I have NOT addressed yet

Being upfront, because it affects one result. The Granian maintainer (u/gi0baro) explained
that drf-granian's memory number comes from me not setting --blocking-threads or
backpressure, so it spawns a lot of threads and burns time on GIL contention - drf-granian's
456MB is that same unfixed issue. He also noted that --cpus=1 in Docker is a time-slice
scheduler limit rather than a real core pin, and that it penalizes Granian more than other
servers since Granian runs I/O in a separate runtime with extra threads.

I'm keeping the CPU cap because it makes runs reproducible and comparable, but he's right
on both counts. Tuning --blocking-threads is top of my list for the next iteration, along with cold
start time and image size (both suggested by the Litestar author).

Takeaways

Framework choice barely matters once you touch I/O. Optimize your queries - the N+1 you didn't notice costs more than any framework migration will ever save.
For raw serialization, Bolt and Litestar lead, then FastAPI, then a big gap.
Cookie JWT auth costs 5-20% on a DB-heavy endpoint; where the token is validated (Rust vs Python, middleware vs dependency) decides where in that range you land.
uvicorn vs granian is workload-dependent: uvicorn wins CPU-bound JSON on ASGI frameworks, granian wins DB-bound endpoints, and granian clearly beats uvicorn for WSGI DRF.
Measure across container restarts and publish the spread - it changed these results more than any framework upgrade did.

Everything - code, Docker setup, runner, raw per-framework JSON results, and dated
snapshots - is in the repo:

https://github.com/huynguyengl99/python-api-frameworks-benchmark

Issues and PRs are very welcome, especially if you know these servers better than I do.
And if you find it useful, a star would be appreciated 😄

Streaming Agents with Pydantic AI over Typed WebSockets

Huy Nguyen — Sat, 08 Aug 2026 14:34:43 +0000

I've been building realtime products for years: group chat, voice pipelines, streaming AI assistants. Two kinds of pain kept following me around.

The first was WebSockets. Every project grew the same untyped receive_json handler with an if/else chain, hand-written validation, and no contract the frontend could rely on. Messages were stringly-typed, bugs surfaced at runtime, and onboarding a teammate meant "read the consumer, end to end". That pain is why I built chanx: typed messages, automatic routing, AsyncAPI docs (the full story is in my intro post).

The second was agent frameworks. I shipped my first LLM features on LangChain and spent more time debugging its abstractions than my own code. Pydantic AI won me over the first time I tried it: typed tools, typed outputs, typed dependency injection. It applies to agents the same philosophy chanx applies to WebSockets, and since both are built on Pydantic they snap together with almost no glue.

There's just one gap. At the UI layer, pydantic-ai's official story is SSE-shaped: adapters for AG-UI and the Vercel AI protocol, and a built-in Agent.to_web() chat page that the docs explicitly label as a dev tool. Real agent products want more than a one-way stream. A conversation is bidirectional: the user sends prompts, the agent streams back text and tool activity, and (the interesting part) the agent sometimes has to stop and ask permission before running a destructive tool. That round trip (pause, ask, resume) fits a WebSocket perfectly, and there is currently no WebSocket example anywhere in the pydantic-ai docs.

So I built one, the same way I build REST APIs: contract-first. The server defines every message as a Pydantic model, chanx turns those models into routed handlers plus an AsyncAPI 3.0 schema, and the React client generates its TypeScript types from that schema. Backend and frontend can't drift, teammates integrate from the contract instead of reading your Python, and the whole thing (streaming, tool calls, approvals) is testable without ever calling an LLM.

Companion repo with everything runnable: huynguyengl99/pydantic-ai-ws-agent. Snippets below are trimmed for reading and current as of pydantic-ai 2.9 and chanx 2.8; the repo is the source of truth.

The big picture

The demo agent ("Tasklet") manages a task list in SQLite. Safe tools run freely; destructive ones pause for approval.

pydantic-ai Agent ──run_stream events──▶ AgentHarness (background task)
      │                                        │ broadcast_event
  typed tools                                  ▼
  requires_approval=True          group: conversation.{id}   ◀── HTTP POST /notify
                                               │ channel layer (in-memory / Redis)
                              ┌────────────────┼────────────────┐
                              ▼                ▼                ▼
                      chanx consumer    chanx consumer   chanx consumer      AsyncAPI 3.0 schema
                        (tab 1)           (tab 2)        (other device)  ──▶ generated TS types

Four pieces: a plain pydantic-ai agent (typed tools, typed deps, requires_approval on the dangerous ones, structured output); a harness that runs it and translates each event into a typed message; a channel layer that carries those events to a per-conversation group rather than a single socket; and a chanx consumer that validates, routes, and documents the protocol the React client generates its types from.

The agent: tools that know they're dangerous

Nothing here is WebSocket-aware. It's a standard pydantic-ai agent with two deliberate choices: a structured output type, and typed deps carrying the conversation id so every tool operates on that conversation's tasks.

class TextAnswer(BaseModel):
    """Final answer to the user, with suggested follow-up prompts."""

    content: str
    follow_ups: list[str] = Field(default_factory=list)


@dataclass
class AgentDeps:
    conversation_id: str


# Every run ends as either a final answer or a request for tool approval.
AgentOutput = TextAnswer | DeferredToolRequests


def build_agent(model: Model | str | None = None) -> Agent[AgentDeps, AgentOutput]:
    agent: Agent[AgentDeps, AgentOutput] = Agent(
        model or settings.resolved_model,
        deps_type=AgentDeps,
        output_type=[TextAnswer, DeferredToolRequests],
        instructions=INSTRUCTIONS,
    )

    @agent.tool
    async def add_task(ctx: RunContext[AgentDeps], title: str) -> db.TaskItem:
        """Add a new task with the given title."""
        return await db.add_task(ctx.deps.conversation_id, title)

    @agent.tool(requires_approval=True)
    async def delete_task(ctx: RunContext[AgentDeps], task_id: int) -> str:
        """Permanently delete the task with the given id. Requires user approval."""
        await db.delete_task(ctx.deps.conversation_id, task_id)
        return f"Task {task_id} deleted"

    return agent

(The repo has six tools; clear_all_tasks is the other one behind approval.)

requires_approval=True is pydantic-ai's deferred tools feature. When the model calls delete_task, the tool does not execute. Instead the run ends early and result.output is a DeferredToolRequests carrying the pending calls, which is exactly why output_type is a union: every run ends as either a final answer or a request for permission, and the type system makes you handle both.

TextAnswer instead of plain str is the structured-output half of the story. Alongside the answer text, the model fills follow_ups with three suggested next prompts that the UI renders as clickable chips. The final answer isn't prose with conventions bolted on; it's a validated object, and the chips ride along in the same schema.

build_agent() being a factory instead of a module-level singleton matters later: tests build the same agent with a fake model.

The message contract

Every WebSocket message is a Pydantic model with a Literal action field. chanx uses that field as a discriminator for validation, routing, and docs:

class ChatMessage(BaseMessage):
    """User sends a prompt to the agent."""
    action: Literal["chat"] = "chat"
    payload: ChatPayload

The full protocol is small. Client → server: chat, tool_decision. Server → client: user_message, stream_start, text_delta, tool_call, tool_result, approval_request, stream_end, suggestions, history, tasks_updated, notification, agent_error.

The consumer binds it together: handlers declare what they accept and what they may reply with, and that declaration is the documentation:

@channel(name="agent", description="Pydantic AI task assistant over a typed WebSocket")
class AgentConsumer(BaseConsumer):
    passthrough_events: ClassVar[list[type[BaseMessage]]] = BROADCAST_MESSAGES

    @ws_handler(
        summary="Send a prompt to the agent",
        output_type=AgentServerMessage,  # the union of everything we may send back
    )
    async def handle_chat(self, message: ChatMessage) -> None:
        self._spawn_run(self.harness.run(message.payload.text))

    @ws_handler(summary="Approve or deny pending tool calls", output_type=AgentServerMessage)
    async def handle_tool_decision(self, message: ToolDecisionMessage) -> None:
        self._spawn_run(self.harness.resolve(message.payload.decisions))

No if/else routing, no manual validation, and /asyncapi.json now serves an AsyncAPI 3.0 spec describing every message above: the WebSocket equivalent of what drf-spectacular gives you for REST (I wrote about that workflow in the schema-first post).

Streaming structured output: snapshots into deltas

Streaming plain text is easy. Streaming a structured answer is the interesting part: the final output is a TextAnswer object, so there is no raw text stream to forward. pydantic-ai's answer is run_stream() plus two hooks.

Tool events arrive through an event_stream_handler, which the harness maps straight onto the message contract:

async def forward_tool_events(
    _ctx: RunContext[AgentDeps], events: AsyncIterable[AgentStreamEvent]
) -> None:
    async for event in events:
        match event:
            case FunctionToolCallEvent(part=part):
                await self.send(ToolCallMessage(payload=ToolCallPayload(
                    tool_call_id=part.tool_call_id,
                    tool_name=part.tool_name,
                    args=part.args_as_dict(),
                )))
            case FunctionToolResultEvent(part=part):
                await self.send(ToolResultMessage(payload=ToolResultPayload(
                    tool_call_id=part.tool_call_id,
                    content=to_jsonable_python(part.content),
                )))

The answer itself streams as partially validated snapshots: stream_output() yields a growing TextAnswer each time more of the model's JSON arrives. Diffing successive content values recovers the text deltas to send:

sent_text = ""
async with self.agent.run_stream(
    user_prompt,
    message_history=history,
    deferred_tool_results=deferred_tool_results,
    deps=AgentDeps(conversation_id=self.conversation_id),
    event_stream_handler=forward_tool_events,
) as stream:
    async for partial in stream.stream_output(debounce_by=None):
        match partial:
            case TextAnswer(content=content) if content and content != sent_text:
                delta = (
                    content.removeprefix(sent_text)
                    if content.startswith(sent_text)
                    else content
                )
                await self.send(TextDeltaMessage(payload=TextDeltaPayload(delta=delta)))
                sent_text = content
        output = partial
    await db.save_history(self.conversation_id, stream.all_messages_json().decode())

The client sees text_delta for prose and tool_call/tool_result pairs for tool activity (rendered as live cards in the UI), and the loop stays completely ignorant of which tools exist.

After the loop, the union output type earns its keep:

match output:
    case DeferredToolRequests() as requests:
        # run_stream stops forwarding events once the final result (the paused
        # calls) is found, so announce them here: same ids, same wire order.
        for part in requests.approvals:
            await self.send(ToolCallMessage(payload=ToolCallPayload(...)))
        PENDING_APPROVALS[self.conversation_id] = requests
        await self.send(approval_request_message(requests))
    case TextAnswer(content=content, follow_ups=follow_ups):
        await self.send(StreamEndMessage(payload=StreamEndPayload(text=content, usage=...)))
        if follow_ups:
            await db.save_suggestions(self.conversation_id, follow_ups)
            await self.send(SuggestionsMessage(payload=SuggestionsPayload(follow_ups=follow_ups)))

One subtlety worth calling out: a deferred tool call never reaches the event_stream_handler (deferred means "not executed", and only executing tools emit events), so the harness announces the paused calls itself before the approval_request. The client renders the same tool cards either way; it can't tell which path produced them. And on the happy path, the suggestions arrive as their own message after stream_end, so the chips pop in once the answer settles.

Broadcast, don't send: background runs and groups

My first version awaited the agent run inside the chat handler and wrote messages straight to the socket. It worked, but with three quiet problems: the handler blocks for the whole run, a page refresh mid-run kills the stream, and a second tab sees nothing.

The fix is the pattern chanx (and Django Channels before it) is built around: broadcast to a group, not to a connection. Every consumer joins a per-conversation group on connect, and the run executes as a background task that broadcasts events to that group:

async def post_authentication(self) -> None:
    group = f"conversation.{self.conversation_id}"
    await self.channel_layer.group_add(group, self.channel_name)
    self.harness = AgentHarness(agent, self.conversation_id, self._broadcast)

async def _broadcast(self, message: BaseMessage) -> None:
    await self.broadcast_event(message, groups=f"conversation.{self.conversation_id}")

def _spawn_run(self, coro) -> None:
    # The task outlives this connection: a client that refreshes mid-run
    # keeps receiving events through the conversation group.
    RUNNING[self.conversation_id] = asyncio.create_task(guarded(coro))

On the receiving side there is nothing to write. chanx's passthrough_events generates the event handlers: any message type in the list that arrives via the group is forwarded to the WebSocket client verbatim, and documented in AsyncAPI like everything else.

What this buys, concretely (all covered by tests):

The chat handler returns immediately; a per-conversation RUNNING dict rejects concurrent runs with a typed agent_error
Refresh-proof streams: the run keeps going without any socket attached; reconnect replays history and picks up live events
Multi-tab for free: every tab renders the same deltas, tool cards, and approval requests. Even the user's own prompt travels this way, echoed as a user_message before stream_start, so the sending tab renders the bubble from the same broadcast as everyone else instead of appending locally
Approve from anywhere: pending approvals are keyed by conversation, not connection; request approval on one device, approve on another

Beyond the socket: HTTP, workers, other frameworks

broadcast_event is a classmethod, so anything can push typed messages into a conversation. The demo exposes POST /conversations/{id}/notify, and a curl fires a toast on every connected tab. The agent uses the same door from the inside: a schedule_reminder tool reads the conversation id from ctx.deps, schedules a job, and returns immediately; when the job fires long after stream_end, it broadcasts a notification into the group.

That escape hatch scales in two directions. Group broadcast isn't the only delivery mode: chanx can address a single connection, so a handler can hand heavy work to a real task queue (taskiq, Celery, ARQ) using self.channel_name as a typed reply address:

# consumer: queue the job with a reply address
await run_agent_job.kiq(prompt, reply_to=self.channel_name)

# worker process: send typed events back to that exact connection
await AgentConsumer.send_event(JobProgressEvent(payload=...), reply_to)

# consumer: typed, routed, and documented like everything else
@event_handler
async def handle_job_progress(self, event: JobProgressEvent) -> ProgressMessage:
    return ProgressMessage(payload=event.payload)

The consumer stays a thin protocol adapter while the LLM work happens in a process built for it. And since the channel layer is just Redis, and chanx speaks the same API on Django Channels and FastAPI, the socket and the agent don't even have to share a service. A split I run in production: Django owns the user-facing WebSocket (sessions, ORM auth, a consumer that's little more than passthrough_events) while a separate FastAPI service runs the models and broadcasts into the same conversation groups. Django never awaits an LLM; the agent service never touches a session. The contract is the only thing they share, and that's the part AsyncAPI documents.

The demo runs on an in-memory layer with zero dependencies; set REDIS_URL and the same code spans processes and instances.

Human-in-the-loop: the approval round trip

This is the part WebSockets were made for. The full sequence:

Model calls delete_task → the run ends with DeferredToolRequests, and the harness announces the paused calls as tool_call messages, so the UI shows the tool card immediately (status: awaiting approval)
The server sends approval_request, persists the message history, and simply stops. No polling, no hanging request
The user clicks Approve or Deny → client sends tool_decision
The harness builds DeferredToolResults and resumes the run with the saved history:

async def resolve(self, decisions: list[ToolDecision]) -> None:
    approvals = {
        d.tool_call_id: (
            ToolApproved(override_args=d.override_args)
            if d.approved
            else ToolDenied(message=d.reason or "Denied by the user")
        )
        for d in decisions
    }
    del PENDING_APPROVALS[self.conversation_id]
    await self.run(None, DeferredToolResults(approvals=approvals))

The resumed run is just another run_stream() call: approved tools execute (their events stream through the same handler), denied tools return the denial message to the model, and the model streams a final response either way. ToolApproved(override_args=...) is a lovely detail: the user can edit the arguments as part of approving, which is how real review UIs should work.

Approvals also survive reconnects, in both directions. Pending requests are keyed by conversation rather than connection, and the consumer re-sends the open approval_request to every new connection, so a refresh mid-approval gets a working approval card instead of a dead end. The E2E test literally approves from a second browser tab. One UI subtlety falls out of multi-tab: when a run resumes, other tabs still showing an open approval card mark it "resolved from another session" on the next stream_end.

The gotcha: tell the model the harness has it covered

My first instructions said "destructive operations require user approval". The model took that personally. Instead of calling delete_task, it asked in chat: "This is permanent, do I have your approval? (yes/no)". Technically obedient, completely bypassing the typed approval flow.

The fix is to be explicit about the division of labor:

When the user asks for a destructive operation, call the tool directly — do NOT
ask for confirmation in chat. The system automatically pauses destructive tool
calls and asks the user for approval in the UI.

If you build approval flows, this prompt bug will find you. The model needs to know that safety is infrastructure, not its job.

AsyncAPI → TypeScript: the client writes itself

chanx serves the contract at /asyncapi.json; every payload above appears in components.schemas as plain JSON Schema. A ~100-line script fetches the spec, collects the client-bound and server-bound messages from the AsyncAPI operations (inputs vs replies), and feeds them to json-schema-to-typescript:

// generated - src/generated/messages.ts
export interface ToolCallMessage {
  action: "tool_call";
  payload: ToolCallPayload;
}
export type ClientMessage = ChatMessage | ToolDecisionMessage;
export type ServerMessage = /* union of all twelve server messages */;

One codegen detail worth knowing: Pydantic marks action as optional in JSON Schema (it has a default), but it's the discriminator, so the script forces it into required for TypeScript to narrow on. With that, the entire client state layer is a single exhaustive switch:

function applyServer(state: ChatState, msg: ServerMessage): ChatState {
  switch (msg.action) {
    case "text_delta":       // append to the streaming bubble
    case "tool_call":        // add a tool card
    case "approval_request": // render Approve/Deny
    case "suggestions":      // show follow-up chips
    case "stream_end":       // finalize + count tokens
    // TypeScript errors if any server message is unhandled
  }
}

Rename a field in a Pydantic payload, run pnpm generate, and tsc lists every client line that just became wrong. Same loop as schema-first REST: the transport changed, the discipline didn't.

Persistence: history as a typed artifact

pydantic-ai ships ModelMessagesTypeAdapter, a TypeAdapter for full message histories, so conversation state is just another validated blob: stream.all_messages_json() on the way out, ModelMessagesTypeAdapter.validate_json(raw) on the way back in. Two functions, no bespoke serialization.

On connect, the consumer replays the conversation as a history message plus the current task list (tasks_updated) and the latest follow-up chips. The transcript builder walks the typed history and reconstructs the interleaving: user text, assistant text, and tool cards in order, each card carrying a done, denied, or awaiting status, so a denied deletion still renders as denied after a refresh. Since everything is stored per conversation, a sidebar with /conversations (titles derived from the first prompt) and conversation delete falls out almost for free. Multi-turn context, reconnects, and resume-after-approval all sit on these same two functions.

Testing the whole flow without an LLM

The part I care most about: the entire WebSocket protocol (streaming, tool execution, approval, denial, history replay) runs in pytest with no API key. pydantic-ai's FunctionModel plays the model; a stream function scripts exactly what "the LLM" does. One wrinkle of structured output: the final answer is itself a call to the output tool, so the script emits that too:

async def delete_task_flow(messages, info):
    tool_return = next(
        (p for p in messages[-1].parts if isinstance(p, ToolReturnPart)), None
    )
    if tool_return is not None:
        # After the tool ran, answer via the structured-output tool
        yield {0: DeltaToolCall(
            name=info.output_tools[0].name,
            json_args=json.dumps({"content": f"Result: {tool_return.content}"}),
        )}
    else:
        yield {0: DeltaToolCall(name="delete_task", json_args='{"task_id": 1}')}

chanx's WebsocketCommunicator drives the consumer, and the assertions read like the protocol spec:

await comm.send_message(ChatMessage(payload=ChatPayload(text="Delete task 1")))
replies = await comm.receive_all_messages(stop_action="approval_request")

# the prompt echoes to the group, the paused call is announced, the run pauses
assert [m.action for m in replies] == [
    "user_message", "stream_start", "tool_call", "approval_request",
]
assert (await db.list_tasks("conv-delete")) != []     # nothing deleted yet

await comm.send_message(ToolDecisionMessage(...approved=True...))
replies = await comm.receive_all_messages(stop_action="stream_end")
assert await db.list_tasks("conv-delete") == []       # now it's gone

The rest of the suite covers the deny path, two tabs receiving identical broadcast feeds, approving from a fresh connection after a disconnect, the concurrent-run guard, transcript replay with denied and awaiting tool cards, and a contract test that pulls /asyncapi.json and asserts every message type is present, so the schema the frontend generates from can't silently lose a message. Twenty tests, a couple of seconds, no LLM.

One trap worth sharing: Agent.override(model=...) uses contextvars, which don't propagate into the consumer's task under the test communicator. The factory pattern sidesteps it: tests call build_agent(FunctionModel(...)) and patch the consumer's agent directly.

Production notes

The demo makes three simplifications you'd revisit for production:

Scaling out: the repo ships a docker-compose.yml for Redis, and the README walks through the two-instance demo (chat on one port, watch the same stream on another).
Auth: the consumer trusts the conversation query param. chanx has an authentication hook (post_authentication) where you'd validate a session or token before joining the group and replaying history.
Durable pending approvals: pending DeferredToolRequests live in an in-process dict; store them in the database if approvals must survive restarts. And if you ever accept message history from the client, run it through pydantic-ai's sanitize_messages first.

Takeaways

WebSockets are the natural transport for agents with approvals: pause and resume are just messages, in both directions, on one connection. pydantic-ai's run_stream() + DeferredToolRequests/DeferredToolResults hand you streaming and human-in-the-loop as library features; the harness adapting them to a transport is ~150 lines
Structured output streams too: partial-validation snapshots diff into text deltas, and extra fields like follow-up suggestions ride along in the same schema
Treat the WebSocket protocol as a contract: Pydantic models with discriminated actions on the server, AsyncAPI in the middle, generated TypeScript on the client
Run in the background, broadcast to a group: refresh-proof streams, multi-tab sync, approve-from-anywhere, and server-initiated notifications all come from one decision; events go to the conversation, not the socket. The same layer lets you offload runs to a worker or put a Django socket in front of a FastAPI agent
Tell the model that approval is handled by infrastructure, or it will improvise its own confirmation dialog in prose
Script the model with FunctionModel and test the entire flow offline; deterministic agent tests are entirely achievable

I hope this saves you some of the pain it took me to learn. If you try the pattern (or have your own tricks for agent WebSockets), the repo issues are open.

🚀 Supercharge Your FastAPI WebSockets with Channel Layers & Group Messaging

Huy Nguyen — Thu, 18 Sep 2025 02:38:11 +0000

Building real-time features in FastAPI? Hit the wall trying to send messages from HTTP endpoints to WebSocket clients? Struggling with group messaging and scalability?

Fast Channels brings Django Channels' battle-tested architecture to FastAPI and the entire ASGI ecosystem.

The Pain Points We've All Hit

FastAPI's native WebSocket support gets you started, but quickly becomes limiting:

❌ No group messaging - can't broadcast to multiple connections

❌ No cross-process communication - HTTP endpoints can't reach WebSocket clients

❌ Manual connection management - tracking users, rooms, cleanup

❌ No scalability - everything breaks with multiple server instances

❌ Testing nightmare - WebSocket unit tests are painful to write

What Fast Channels Delivers

✅ Group messaging - broadcast to chat rooms, user segments, notification groups

✅ Cross-process communication - HTTP → WebSocket, background workers → clients

✅ Auto connection management - join/leave groups automatically

✅ Multi-instance scaling - Redis-backed message routing

✅ Built-in testing framework - proper WebSocket test utilities

✅ Consumer patterns - WebSocket handlers that feel like FastAPI views

Battle-Tested in Production

This isn't experimental tech. Django Channels has powered real-time features for millions of users across thousands of production applications for years. Fast Channels brings that same proven architecture to FastAPI.

# Simple but powerful
class ChatConsumer(AsyncWebsocketConsumer):
    groups = ["chat_room"]  # Auto-join on connect

    async def receive(self, text_data=None, **kwargs):
        # Broadcast to all users in the room
        await self.channel_layer.group_send("chat_room", {
            "type": "chat_message",
            "message": text_data
        })

Production-Ready Features

Redis Integration: Full Redis support with Sentinel for high availability

Multiple Backends: In-memory for testing, Redis Queue for reliability, Redis Pub/Sub for speed

Type Safety: Complete mypy and pyright support

ASGI Universal: Works with FastAPI, Starlette, Quart, any ASGI framework

Testing Built-in: Comprehensive WebSocket testing utilities

FastAPI vs Fast Channels vs Alternatives

Feature	Native FastAPI	Broadcaster	Fast Channels
Basic WebSocket	✅	✅	✅
Group messaging	❌	✅	✅
Consumer patterns	❌	❌	✅
Message persistence	❌	❌	✅
Testing framework	❌	❌	✅
Auto connection mgmt	❌	❌	✅
Cross-process msgs	❌	❌	✅
Production battle-tested	❌	❌	✅

Real-World Impact

Teams are using Fast Channels for:

🏪 E-commerce: Live inventory, order tracking, flash sale updates

💬 Chat platforms: Real-time messaging, typing indicators, presence

📊 Dashboards: Live metrics, system monitoring, alerts

🎮 Gaming: Multiplayer sync, leaderboards, real-time events

🔔 Notifications: Push alerts, system updates, user notifications

Quick Start

pip install fast-channels[redis]

Register a channel layer, create a consumer, add to FastAPI - done. Your WebSockets now scale across multiple processes and servers.

Why It Matters

Real-time features shouldn't require reinventing messaging infrastructure. Fast Channels gives you Django's proven patterns with FastAPI's developer experience - the best of both worlds.

Production-ready: Redis Sentinel support, proper error handling, connection cleanup

Developer-friendly: Type safety, testing utilities, clear documentation

Scalable: Multi-process, multi-server, battle-tested architecture

Stop building WebSocket boilerplate. Start building features.

🔗 Links:

Building real-time features? What's your biggest WebSocket challenge? 👇

Building a Modern, Type-Safe Authentication System for Django REST Framework

Huy Nguyen — Wed, 09 Jul 2025 13:59:32 +0000

Have you ever spent hours fighting with authentication setup in your Django REST API? Or struggled with type errors that could have been caught at development time? Or wished your API documentation would just... work automatically?

I've been there. After years of wrestling with existing Django authentication packages, I decided to build something better: DRF Auth Kit – a modern, type-safe authentication toolkit that just works.

The Problem: Authentication Shouldn't Be This Hard

Let me paint a picture. You're building a modern web app with a Django REST API backend. You need:

✅ JWT authentication with cookie support

✅ Multi-factor authentication

✅ Social login (Google, GitHub, etc.)

✅ Automatic API documentation

✅ Type safety throughout

With existing solutions, you'd typically:

Install 3-4 different packages
Write custom serializers and views
Manually configure complex URL patterns
Struggle with type hints and documentation
Debug cryptic authentication errors

There had to be a better way.

The Solution: DRF Auth Kit

DRF Auth Kit is what I wish existed when I started building Django APIs. Here's what makes it different:

🎯 One Package, Everything Included

# Get everything you need
pip install drf-auth-kit[all]

# Or pick what you need
pip install drf-auth-kit[mfa]     # + Multi-factor auth
pip install drf-auth-kit[social]  # + Social authentication

🔒 Universal Authentication Backend

Instead of juggling multiple authentication classes:

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'auth_kit.authentication.AuthKitAuthentication',
    ],
}

AUTH_KIT = {
    'AUTH_TYPE': 'jwt',  # or 'token', 'custom'
    'USE_AUTH_COOKIE': True,
    'USE_MFA': True,
}

This single configuration handles:

JWT tokens with automatic refresh
DRF token authentication
Custom authentication backends
HTTP-only cookies with security headers
Multi-factor authentication

🛠️ Intelligent URL Management

Conditional URL patterns - URLs automatically included/excluded based on settings:

# When USE_MFA=True, automatically creates:
# POST /api/auth/login/         # First step (password)
# POST /api/auth/login/verify/  # Second step (MFA code)
# POST /api/auth/login/change-method/
# POST /api/auth/login/resend/
# GET /api/auth/mfa/           # MFA management endpoints

# When USE_MFA=False, creates:
# POST /api/auth/login/         # Direct login

Social authentication - automatic endpoint generation:

# Just add a provider to INSTALLED_APPS:
INSTALLED_APPS = [
    'allauth.socialaccount.providers.google',
    'auth_kit.social',
]

# Automatically creates:
# POST /api/auth/social/google/
# POST /api/auth/social/google/connect/
# GET /api/auth/social/accounts/

📝 Type Safety That Actually Works

Every component includes comprehensive type hints:

from auth_kit.authentication import AuthKitAuthentication
from auth_kit.serializers import LoginRequestSerializer
from auth_kit.views import LoginView

Full mypy and pyright compatibility with comprehensive type hints throughout the codebase. Your IDE will thank you. No more guessing what parameters a function expects or what a setting does.

📚 API Documentation That Updates Itself

Integration with DRF Spectacular means your API docs are always current:

# Zero configuration needed
INSTALLED_APPS = [
    'drf_spectacular',
    'auth_kit',
]

Visit /api/docs/ and you'll see:

Interactive Swagger UI
Complete request/response schemas
Authentication flow examples
Working test interface

🛡️ MFA Made Simple

Multi-factor authentication that doesn't require a PhD:

# Enable MFA
INSTALLED_APPS = [
    'auth_kit',
    'auth_kit.mfa',  # Just add this!
]

AUTH_KIT = {
    'USE_MFA': True,  # URLs automatically reconfigure
}

Automatic URL reconfiguration when MFA is enabled:

Login endpoint becomes two-step flow
MFA management endpoints automatically created
All endpoints documented in your OpenAPI schema

You get:

Email-based MFA with HTML/text templates
Authenticator app support (Google Authenticator, Authy)
Backup codes for account recovery
Extensible system for custom MFA methods

The login flow becomes a clean two-step process:

Username/password → Get ephemeral token
MFA code → Get final auth tokens

🌐 Social Authentication That Just Works

Social login with 50+ providers and automatic URL generation:

INSTALLED_APPS = [
    'auth_kit',
    'auth_kit.social',  # Automatic URL generation
    'allauth.socialaccount.providers.google',
    'allauth.socialaccount.providers.github',
    # Add any provider, URLs created automatically!
]

# URLs are automatically created for each provider:
# POST /api/auth/social/google/
# POST /api/auth/social/github/
# POST /api/auth/social/google/connect/
# POST /api/auth/social/github/connect/
# GET /api/auth/social/accounts/

No manual URL configuration needed! The system scans your installed providers and creates endpoints automatically.

🍪 Security-First Cookie Authentication

HTTP-only cookies with all the security features:

AUTH_KIT = {
    'USE_AUTH_COOKIE': True,
    'AUTH_COOKIE_SECURE': True,      # HTTPS only
    'AUTH_COOKIE_HTTPONLY': True,    # No JavaScript access
    'AUTH_COOKIE_SAMESITE': 'Lax',   # CSRF protection
}

Perfect for modern web apps where you want security without complexity.

Real-World Example: Complete Auth Setup

Here's how you'd set up a complete authentication system:

# settings.py
INSTALLED_APPS = [
    'rest_framework',
    'auth_kit',
    'auth_kit.mfa',
    'auth_kit.social',
    'allauth',
    'allauth.account',
    'allauth.socialaccount',
    'allauth.socialaccount.providers.google',
]

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'auth_kit.authentication.AuthKitAuthentication',
    ],
}

AUTH_KIT = {
    'AUTH_TYPE': 'jwt',
    'USE_AUTH_COOKIE': True,
    'USE_MFA': True,
}

# Social auth providers
SOCIALACCOUNT_PROVIDERS = {
    'google': {
        'SCOPE': ['profile', 'email'],
        'AUTH_PARAMS': {'access_type': 'online'},
        'OAUTH_PKCE_ENABLED': True,
        'APP': {
            'client_id': 'your-google-client-id',
            'secret': 'your-google-client-secret',
        }
    }
}

# urls.py
from django.urls import path, include

urlpatterns = [
    path('api/auth/', include('auth_kit.urls')),
    path('api/auth/social/', include('auth_kit.social.urls')),
]

That's it! You now have:

JWT authentication with cookies
User registration and email verification
Password reset functionality
Multi-factor authentication
Social login with Google
Complete API documentation
Type safety throughout

Customization: Make It Yours

Need custom behavior? Override any component:

# Custom login serializer
class CustomLoginSerializer(LoginRequestSerializer):
    remember_me = serializers.BooleanField(default=False)

    def validate(self, attrs):
        attrs = super().validate(attrs)
        # Your custom logic here
        return attrs

# Apply it
AUTH_KIT = {
    'LOGIN_REQUEST_SERIALIZER': 'myapp.serializers.CustomLoginSerializer',
}

Want a custom MFA method? Easy:

class SMSMFAHandler(BaseMFAHandler):
    name = "sms"

    def send_code(self, user: User, code: str) -> None:
        # Your SMS implementation
        send_sms(user.phone, f"Your verification code: {code}")

# Register it
AUTH_KIT = {
    'MFA_HANDLERS': [
        'auth_kit.mfa.handlers.app.MFAAppHandler',
        'auth_kit.mfa.handlers.email.MFAEmailHandler',
        'myapp.handlers.SMSMFAHandler',
    ],
}

Production-Ready Features

DRF Auth Kit isn't just for prototypes:

Comprehensive test coverage (>95%) across multiple Python/Django versions
Internationalization support for 57 languages
Security best practices built-in
Performance optimized with efficient database queries
Rate limiting and audit logging (coming soon)

The Numbers Don't Lie

Here's how DRF Auth Kit compares to existing solutions:

Feature	dj-rest-auth	django-trench	DRF Auth Kit
Type Safety	❌	❌	✅ Full mypy/pyright
Auto OpenAPI	⚠️ Partial	❌	✅ Complete
Cookie Auth	✅	❌	✅ Enhanced
MFA Support	❌	✅	✅ Simplified
Social Auth	✅	❌	✅ Better integration
Custom Auth	⚠️ Complex	❌	✅ Easy
I18n Support	❌	❌	✅ 57 languages
Test Coverage	⚠️	⚠️	✅ >95%

Try It Today

pip install drf-auth-kit[all]

Check out the complete documentation for detailed guides, examples, and API reference.

What's Next?

The roadmap includes exciting features:

WebAuthn support for passwordless authentication
Hardware security keys (YubiKey, FIDO2)
Rate limiting and audit logging
Advanced security features

Join the Community

DRF Auth Kit is open source and built by developers, for developers:

GitHub: github.com/forthecraft/drf-auth-kit
Documentation: drf-auth-kit.readthedocs.io
PyPI: pypi.org/project/drf-auth-kit

Final Thoughts

Building authentication shouldn't be a painful experience. With DRF Auth Kit, you get a modern, type-safe solution that grows with your project – from simple JWT authentication to complex multi-factor flows.

The combination of type safety, automatic documentation, and developer-friendly APIs makes it perfect for both new projects and existing ones looking to modernize their authentication.

What are you waiting for? Give it a try in your next project and let me know what you think! 🚀

Chanx: The Missing Toolkit for Django Channels - Production-Ready WebSockets Made Simple

Huy Nguyen — Wed, 11 Jun 2025 08:13:18 +0000

Building WebSocket applications with Django Channels just got a whole lot easier

The Problem with Django Channels

Don't get me wrong - Django Channels is fantastic for adding WebSocket support to Django applications. But after building several real-time apps, I kept running into the same challenges:

Setting up authentication for WebSocket connections
Validating message structures and preventing runtime errors
Testing multi-user WebSocket interactions
Managing group messaging patterns
Debugging WebSocket connections during development

I found myself writing the same boilerplate code over and over. So I built Chanx to solve these problems once and for all.

What is Chanx?

Chanx is a comprehensive toolkit that extends Django Channels with production-ready features. Think of it as what Django REST Framework is to Django views - but for WebSocket consumers.

Key Features That'll Save You Hours

🔐 DRF-Style Authentication

Use your existing authentication classes with WebSockets:

from chanx.generic.websocket import AsyncJsonWebsocketConsumer
from rest_framework.authentication import SessionAuthentication
from rest_framework.permissions import IsAuthenticated

class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessage]):
    authentication_classes = [SessionAuthentication]
    permission_classes = [IsAuthenticated]

    async def receive_message(self, message: ChatMessage, **kwargs):
        # Only authenticated users reach here
        # self.user is automatically populated
        pass

📝 Type-Safe Messaging with Pydantic

No more runtime errors from malformed messages:

from typing import Literal
from chanx.messages.base import BaseMessage

class ChatMessage(BaseMessage):
    action: Literal["chat"] = "chat"
    payload: str

class NotificationMessage(BaseMessage):
    action: Literal["notify"] = "notify"
    payload: dict

# Union type for type safety
AllMessages = ChatMessage | NotificationMessage

class MyConsumer(AsyncJsonWebsocketConsumer[AllMessages]):
    async def receive_message(self, message: AllMessages, **kwargs):
        # Pattern matching with full type safety
        match message:
            case ChatMessage(payload=text):
                await self.handle_chat(text)
            case NotificationMessage(payload=data):
                await self.handle_notification(data)

🎮 WebSocket Playground

Like Swagger, but for WebSockets! Auto-discovers your endpoints and lets you test them interactively:

# settings.py
INSTALLED_APPS = [
    # ...
    'chanx.playground',
]

# urls.py
urlpatterns = [
    path('playground/', include('chanx.playground.urls')),
]

Visit /playground/websocket/ and test your WebSocket endpoints without writing JavaScript!

👥 Simplified Group Messaging

Pub/sub messaging made easy:

class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessage]):
    async def build_groups(self) -> list[str]:
        room_id = self.scope["url_route"]["kwargs"]["room_id"]
        return [f"chat_room_{room_id}"]

    async def receive_message(self, message: ChatMessage, **kwargs):
        # Broadcast to all users in the room
        await self.send_group_message(message)

Messages automatically include metadata:

{
  "action": "chat",
  "payload": "Hello everyone!",
  "is_mine": false,
  "is_current": false
}

🧪 Testing That Actually Works

Multi-user WebSocket testing without the headaches:

from chanx.testing import WebsocketTestCase

class ChatTest(WebsocketTestCase):
    ws_path = "/ws/chat/room1/"

    async def test_multi_user_chat(self):
        # First user (automatic setup)
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Second user with different credentials
        user2_comm = self.create_communicator(headers=user2_headers)
        await user2_comm.connect()

        # Test group messaging
        await self.auth_communicator.send_message(ChatMessage(payload="Hello!"))
        responses = await user2_comm.receive_all_json(wait_group=True)

        assert responses[0]["payload"] == "Hello!"
        assert responses[0]["is_mine"] == False

Real-World Example: Building a Chat App

Here's how simple it is to build a production-ready chat application:

1. Define Your Messages

from typing import Literal
from chanx.messages.base import BaseMessage, BaseGroupMessage

class ChatMessage(BaseMessage):
    action: Literal["chat"] = "chat"
    payload: str

class UserJoinedMessage(BaseGroupMessage):
    action: Literal["user_joined"] = "user_joined"
    payload: dict

ChatMessages = ChatMessage | PingMessage  # Include ping for health checks

2. Create Your Consumer

from chanx.generic.websocket import AsyncJsonWebsocketConsumer
from chanx.messages.incoming import PingMessage
from chanx.messages.outgoing import PongMessage

class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessages]):
    authentication_classes = [SessionAuthentication]
    permission_classes = [IsAuthenticated]

    async def build_groups(self) -> list[str]:
        room_id = self.scope["url_route"]["kwargs"]["room_id"]
        return [f"chat_room_{room_id}"]

    async def post_authentication(self):
        """Called after successful authentication"""
        await self.send_group_message(
            UserJoinedMessage(payload={"username": self.user.username})
        )

    async def receive_message(self, message: ChatMessages, **kwargs):
        match message:
            case PingMessage():
                await self.send_message(PongMessage())
            case ChatMessage(payload=text):
                # Save to database and broadcast
                await self.save_message(text)
                await self.send_group_message(message)

3. Set Up Routing

# chat/routing.py
from chanx.routing import path
from channels.routing import URLRouter

router = URLRouter([
    path("<int:room_id>/", ChatConsumer.as_asgi()),
])

# project/routing.py
from chanx.routing import include, path
from channels.routing import URLRouter

router = URLRouter([
    path("ws/chat/", include("chat.routing")),
])

4. Test It

class ChatRoomTest(WebsocketTestCase):
    ws_path = "/ws/chat/123/"

    async def test_user_can_join_and_chat(self):
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Should receive user_joined message
        messages = await self.auth_communicator.receive_all_json()
        assert any(msg["action"] == "user_joined" for msg in messages)

        # Send a chat message
        await self.auth_communicator.send_message(ChatMessage(payload="Hello!"))
        responses = await self.auth_communicator.receive_all_json(wait_group=True)

        assert responses[0]["action"] == "chat"
        assert responses[0]["payload"] == "Hello!"

Why Developers Love Chanx

Complete Type Safety

# Generic type parameters for compile-time checking
class MyConsumer(AsyncJsonWebsocketConsumer[
    IncomingMessages,  # Required: Your message types
    ChannelEvents,     # Optional: Channel layer events
    Room              # Optional: Model for object permissions
]):
    queryset = Room.objects.all()  # For object-level permissions

Enhanced Routing

Django-style routing specifically designed for WebSockets:

from chanx.routing import path, re_path, include

# Use familiar Django patterns
router = URLRouter([
    path("room/<str:room_name>/", ChatConsumer.as_asgi()),
    re_path(r"^admin/(?P<id>\d+)/$", AdminConsumer.as_asgi()),
    path("api/", include("api.routing")),
])

Channel Events

Type-safe communication between consumers:

# Send events from views, tasks, or other consumers
ChatConsumer.send_channel_event(
    "chat_room_123",
    NotificationEvent(payload={"message": "System update"})
)

# Handle in consumer
async def receive_event(self, event: NotificationEvent):
    match event:
        case NotificationEvent():
            await self.send_message(SystemMessage(payload=event.payload))

Getting Started

pip install chanx

Essential links:

📚 Documentation - Comprehensive guides and examples
💻 GitHub - Source code and issues
🚀 Complete Example - Production-ready chat app

Quick setup:

# settings.py
INSTALLED_APPS = [
    'channels',
    'rest_framework', 
    'chanx.playground',  # For WebSocket testing UI
]

CHANX = {
    'SEND_COMPLETION': True,  # Important for testing
    'SEND_AUTHENTICATION_MESSAGE': True,
}

Perfect for AI Applications

Chanx is especially powerful for AI chatbots and streaming applications:

async def handle_ai_chat(self, user_message: str):
    """Stream AI response token by token"""
    async for token in self.get_ai_stream(user_message):
        await self.send_message(AIStreamingMessage(payload=token))

Community and Feedback

Chanx is actively maintained and used in production applications. The community is growing, and I'm always looking for feedback:

What challenges are you facing with Django Channels?
Which features would be most valuable to you?
How can we make WebSocket development even easier?

Conclusion

Django Channels is powerful, but building production WebSocket applications shouldn't require reinventing the wheel every time. Chanx provides the missing pieces:

✅ Authentication - DRF integration out of the box

✅ Type Safety - Catch errors at development time

✅ Testing - Multi-user scenarios made simple

✅ Developer Tools - WebSocket playground for debugging

✅ Group Messaging - Pub/sub patterns simplified

✅ Production Ready - Battle-tested in real applications

Whether you're building a chat application, real-time collaboration tool, or AI-powered interface, Chanx helps you ship faster with fewer bugs.

Try it out and let me know what you think! 🚀

What's your experience with Django Channels? Have you run into similar challenges? Share your thoughts in the comments!

🚀 Streamline Your Python & Django Development with Production-Ready Cookiecutter Templates

Huy Nguyen — Mon, 09 Jun 2025 15:18:53 +0000

TL;DR: I've created a comprehensive cookiecutter template collection that generates production-ready Python apps, PyPI packages, Django REST APIs, and DRF packages with modern tooling, best practices, and zero configuration hassle.

The Problem: Starting Projects Shouldn't Be This Hard

How many times have you started a new Python project and spent hours setting up the same boilerplate code? Configuring pyproject.toml, setting up pre-commit hooks, writing GitHub Actions, configuring Docker, setting up Django with all the modern best practices...

The list goes on. And then you realize you forgot to configure type checking, or your test coverage setup is broken, or you're missing that one crucial piece of tooling that makes development smooth.

I've been there. We've all been there.

The Solution: One Template to Rule Them All (Actually, Four)

That's why I created cookiecutter-python — a curated collection of four specialized cookiecutter templates that cover the most common Python development scenarios:

🐍 Python App - For General-Purpose Applications

Perfect for CLI tools, data processing scripts, or any standalone Python application.

📦 Python Package - For PyPI Publishing

Ready-to-publish package structure with documentation, CI/CD, and multiple license options.

🌐 Django REST App - For Full-Stack Web APIs

Complete Django REST Framework setup with authentication, database, optional Celery/WebSockets, and cloud deployment support.

🧩 DRF Package - For Reusable Django Components

Lightweight structure for building and distributing Django REST Framework extensions.

What Makes These Templates Special?

✨ Modern Tooling Out of the Box

Each template is carefully configured with specialized tooling for its use case:

🔧 Universal Tools (All Templates):

UV for lightning-fast dependency management with intelligent dependency groups
Ruff for blazing-fast linting and formatting
Black for consistent code formatting
Dual type checking with both mypy and pyright for maximum type safety
Pre-commit hooks for automated quality checks
pytest with parallel execution (pytest-xdist) and coverage reporting
GitHub Actions CI/CD pipelines

📦 Package Templates (PyPackage & DRF Package) Add:

Commitizen for conventional commits and automated versioning
Interrogate for docstring coverage monitoring with badge generation
Sphinx ecosystem for professional documentation (myst-parser, ReadTheDocs theme)
Automated publishing workflows to PyPI

🌐 Django Templates (DRF & DRF Package) Add:

Factory Boy for realistic test data generation
Faker for generating fake data
django-stubs + djangorestframework-stubs for comprehensive Django type checking
django-debug-toolbar for development debugging
IPython for enhanced Django shell experience
Freezegun for time-based testing

🏗️ Production-Ready Architecture

The Django templates include:

PostgreSQL integration with connection pooling
Redis support for Celery and WebSockets
JWT authentication with django-rest-auth
API documentation with Swagger/ReDoc
Docker containerization
Cloud storage support (AWS S3, GCP, Azure)
Email providers integration (SendGrid, Mailgun, etc.)

🎯 Zero Configuration Philosophy

Each generated project is immediately runnable. No additional setup required. Just generate, install dependencies, and start developing.

Quick Start Guide

Prerequisites

First, you'll need to install cookiecutter. I recommend using Homebrew (macOS) or pipx for a clean installation:

macOS (Recommended):

brew install cookiecutter

Linux/Windows with pipx:

pipx install cookiecutter

Alternative (with pip):

python3 -m pip install --user cookiecutter

Why not just pip install cookiecutter? Since we'll be using UV for Python dependency management in the generated projects, keeping cookiecutter isolated prevents potential conflicts and keeps your global Python environment clean.

Generate Your Project

cookiecutter gh:huynguyengl99/cookiecutter-python

You'll be prompted to select your template type:

Select a template
1 - Python app (A cookiecutter template for a normal python app)
2 - Python package (A cookiecutter template for a python package)  
3 - DRF (A cookiecutter template for a django rest framework app)
4 - DRF package (A cookiecutter template for a drf package)
Choose from [1/2/3/4] (1):

Start Developing Immediately

For a Django REST App:

cd your-project
uv sync                    # Install dependencies
docker compose up -d       # Start PostgreSQL/Redis
scripts/manage.sh migrate  # Run migrations
scripts/manage.sh runserver # Start development server

Your API documentation is immediately available at:

Swagger UI: http://localhost:8000/api/schema/swg/
ReDoc: http://localhost:8000/api/schema/redoc/

Deep Dive: What Each Template Provides

🐍 Python App Template

Perfect for: CLI tools, data processing, automation scripts

Includes:

Modern pyproject.toml configuration
Docker setup for easy deployment
Comprehensive testing framework
Code quality tools (Ruff, Black, mypy, pyright)
GitHub Actions for CI/CD

Generated Structure:

my-app/
├── pyproject.toml
├── docker-compose.yml
├── Dockerfile
├── scripts/
│   └── lint.sh
└── my_app/
    └── __init__.py

📦 Python Package Template

Perfect for: Open-source libraries, reusable components

Includes:

PyPI-ready project structure
Sphinx documentation with ReadTheDocs integration
Multiple license options (MIT, BSD, Apache, GPL)
Automated versioning with Commitizen
Publishing workflow to PyPI

Generated Structure:

my-package/
├── pyproject.toml
├── LICENSE
├── docs/
├── tests/
├── .github/workflows/
└── my_package/
    └── __init__.py

🌐 Django REST App Template

Perfect for: Web APIs, SaaS backends, microservices

Configuration Options:

Database: PostgreSQL with multiple version options
Authentication: JWT with optional Google OAuth
Background Tasks: Optional Celery integration
Real-time: Optional WebSocket support with Channels
Cloud Storage: AWS S3, Google Cloud, Azure support
Email: 10+ provider integrations (SendGrid, Mailgun, etc.)
API Style: Optional camelCase conversion

Generated Structure:

my-api/
├── pyproject.toml
├── docker-compose.yml
├── Dockerfile
├── scripts/
│   └── manage.sh
└── my_api/
    ├── config/
    │   ├── settings/
    │   ├── urls.py
    │   └── wsgi.py
    ├── accounts/        # Custom user model
    ├── core/           # Utilities
    └── status/         # Health check

Key Features:

Custom User Model with email authentication
JWT Authentication with refresh token rotation
API Versioning ready structure
Comprehensive Testing with factories and fixtures
Admin Interface customization
Structured Logging with structured logs
Health Checks and monitoring endpoints

🧩 DRF Package Template

Perfect for: Django REST Framework extensions, reusable API components

Includes:

Sandbox Django project for testing
Documentation with Django integration
Package publishing to PyPI
Testing across multiple Django/DRF versions

Unique Feature: The sandbox directory serves dual purposes:

Testing Environment: Run tests against your package
Development Playground: Manual testing with a real Django app

Advanced Features That Set These Apart

🔧 Intelligent Dependency Management with UV

Each template uses UV's dependency groups for optimal development experience:

[dependency-groups]
dev = ["commitizen", "factory-boy", "pytest-django"]  # Core development
lint = ["black", "ruff", "mypy", "pyright"]           # Code quality  
test = ["pytest-cov", "pytest-xdist", "coverage"]    # Testing tools
docs = ["sphinx", "myst-parser", "interrogate"]      # Documentation

Smart defaults with uv sync:

Application templates: Auto-installs dev + lint groups
Package templates: Auto-installs all groups for comprehensive development
Selective installation: uv sync --group test for specific workflows

🛡️ Type Safety First

Every template includes comprehensive type checking with dual tools for maximum coverage:

mypy with specialized Django and DRF plugins for framework-specific type checking
pyright for advanced static analysis and fast feedback
django-stubs for comprehensive Django typing (Django templates)

This dual approach catches different categories of type issues and provides the most robust type safety available in the Python ecosystem.

📊 Quality Metrics & Automation

Package templates include sophisticated quality monitoring:

# Docstring coverage with automatic badge generation
interrogate --generate-badge docs/_static/ my_package

# Conventional commits with automated versioning
cz commit  # Interactive commit creation
cz bump    # Automatic version bumping and changelog

# Parallel testing for faster feedback
pytest -n5  # Runs tests across 5 parallel processes

Quality gates enforce standards:

80% docstring coverage requirement
100% type hint coverage with mypy strict mode
Zero linting violations with Ruff and Black
Conventional commit format for automated changelog generation

⚡ Modern Python Practices

PEP 621 compliant pyproject.toml
UV for dependency management (10x faster than pip)
Ruff for linting (100x faster than flake8)
Python 3.10+ with modern syntax support

🚀 Production Deployment Ready

Django templates include:

Multi-stage Docker builds for optimized images
Gunicorn configuration for production WSGI
Environment-based settings (dev/test/production)
Static file handling with WhiteNoise
Database connection pooling

Real-World Usage Examples

Scenario 1: Building a SaaS API

cookiecutter gh:huynguyengl99/cookiecutter-python
# Select: 3 - DRF
# Configure: PostgreSQL, JWT auth, Celery, SendGrid emails, AWS S3

cd my-saas-api
uv sync
docker compose up -d
scripts/manage.sh migrate
scripts/manage.sh createsuperuser
scripts/manage.sh runserver

Result: Full-featured API with user authentication, background tasks, email integration, and cloud storage. Ready for production deployment.

Scenario 2: Publishing a Python Library

cookiecutter gh:huynguyengl99/cookiecutter-python
# Select: 2 - Python package
# Configure: MIT license, GitHub username, PyPI publishing

cd my-library
uv sync
pytest  # Tests pass immediately
scripts/lint.sh  # Code quality checks pass
cz commit  # Conventional commits

Result: Professional package structure ready for PyPI publishing with documentation, CI/CD, and automated releases.

Scenario 3: Building a DRF Extension

cookiecutter gh:huynguyengl99/cookiecutter-python
# Select: 4 - DRF package

cd my-drf-extension
uv sync
docker compose up -d  # Test database
pytest sandbox  # Test against real Django app
python sandbox/manage.py runserver  # Manual testing

Result: Professional DRF package with testing sandbox and publishing workflow.

Best Practices Baked In

🧪 Testing Excellence

Each template includes sophisticated testing infrastructure:

🏃‍♂️ Performance-Optimized Testing:

Parallel execution with pytest-xdist (5 parallel processes by default)
Smart test discovery with proper Django integration
Coverage reporting with branch coverage and XML output for CI integration

🏭 Test Data Generation (Django Templates):

# Factory Boy integration for realistic test data
from accounts.factories import UserFactory

def test_user_creation():
    user = UserFactory.create(email="test@example.com")
    assert user.is_active

⏰ Time-Based Testing:

# Freezegun for testing time-dependent code
from freezegun import freeze_time

@freeze_time("2023-01-01")
def test_timestamp_behavior():
    # Test behavior at specific time
    pass

🔧 Advanced Testing Features:

pytest-mock for sophisticated mocking
pytest-django for Django-specific fixtures and database handling
Factory Boy with custom factory metaclass for type-safe model factories
Coverage exclusions for migrations, settings, and other non-testable code

📝 Documentation Standards

Package templates include professional documentation tooling:

📖 Sphinx Documentation:

Automatic API documentation generation
ReadTheDocs integration with automated builds
MyST parser for Markdown support in reStructuredText projects
Cross-references and intersphinx linking

📊 Documentation Quality Monitoring:

Interrogate for docstring coverage monitoring
Automatic badge generation showing documentation coverage percentage
80% coverage requirement enforced in CI/CD
SVG badges integrated into documentation and README

# Check docstring coverage
interrogate -vv my_package

# Generate coverage badge
interrogate --generate-badge docs/_static/

📋 API Documentation (Django Templates):

Swagger UI and ReDoc auto-generation from code
OpenAPI 3.0 specification export
Interactive API testing directly in documentation

🔄 CI/CD Pipeline

GitHub Actions workflows include:

🧪 Comprehensive Testing:

Multi-version testing across Python 3.10-3.13
Parallel test execution with pytest-xdist for faster feedback
Coverage reporting with Codecov integration and XML output
Matrix testing across multiple dependency versions (Django templates)

✅ Code Quality Gates:

Linting pipeline with Ruff for import sorting and style checks
Dual type checking with both mypy and pyright
Code formatting verification with Black
TOML file validation and sorting with toml-sort

📦 Automated Publishing (Package Templates):

Semantic versioning with Commitizen integration
Automated PyPI publishing on tagged releases
GitHub releases with automated changelog generation
Build artifact caching for faster CI runs

🔒 Security & Quality:

Dependency vulnerability scanning
Pre-commit hook validation in CI
Build reproducibility with UV lock files

🎨 Code Quality

Pre-commit hooks automatically run:

Ruff for linting and import sorting
Black for code formatting
toml-sort for TOML file organization
Type checking with mypy/pyright

Common Questions

Q: Why four separate templates instead of one configurable template?

A: Each template is optimized for its specific use case. A Python package has different needs than a Django API. Separate templates mean:

Cleaner code without excessive conditionals
Faster generation with only relevant files
Better maintenance with focused concerns
Easier customization for specific needs

Q: Why UV instead of pip/poetry?

A: UV is incredibly fast (10-100x faster than pip) and handles both Python version management and dependency resolution. It's the future of Python package management.

Q: Can I customize the generated templates?

A: Absolutely! The templates are designed to be starting points. All configuration is in standard files (pyproject.toml, settings.py, etc.) that you can modify as needed.

Q: What about database migrations in Django templates?

A: The Django templates include a custom user model and initial migrations. After generation, run scripts/manage.sh migrate and you're ready to go.

Future Roadmap

I'm continuously improving these templates based on community feedback:

FastAPI template for modern async APIs
Next.js + Django full-stack template
Kubernetes deployment configurations
More cloud provider integrations
Additional authentication providers

Getting Started Today

Ready to streamline your Python development? Here's how to get started:

Install cookiecutter: pip install cookiecutter
Generate your project: cookiecutter gh:huynguyengl99/cookiecutter-python
Choose your template and configure options
Start developing immediately!

Useful Links

📂 Repository: github.com/huynguyengl99/cookiecutter-python
📖 Documentation: Full README with detailed setup instructions
🐛 Issues: Report bugs or request features
💡 Discussions: Share your use cases and get help

Conclusion

These cookiecutter templates represent hundreds of hours of research, testing, and refinement to create the ideal starting point for Python projects. Whether you're building a simple CLI tool or a complex Django API, you can focus on your core business logic instead of wrestling with configuration.

The templates embody modern Python best practices:

Type safety with comprehensive type checking
Code quality with automated formatting and linting
Testing excellence with comprehensive test setups
Production readiness with Docker and deployment configs
Developer experience with helpful scripts and documentation

Give them a try for your next project. I'm confident they'll save you hours of setup time and help you ship better code faster.

What are you waiting for? Let's build something amazing together! 🚀

Have you used these templates? Found them helpful? Have suggestions for improvements? I'd love to hear about your experience! Drop a comment below or open an issue on GitHub.

Easy Multilingual Page (Scales to Over 50 Languages)

Huy Nguyen — Fri, 21 Jun 2024 14:17:13 +0000

Hi all,

So, a few months back, I kicked off a project to create a multilingual website. Let me tell you, managing content in multiple languages is a real headache. It’s doable if you’re only dealing with 3 or 4 languages, but once you hit 30+, it’s a whole different ball game trying to keep everything updated and consistent.

To make life easier, I decided to bring in a CMS for my Astro-based site. I dug into frameworks like Wagtail and Strapi since I’m familiar with Python and Node.js, but man, those were tough nuts to crack. Plus, I wanted a centralized place for all my content instead of having to jump between different pages for each language. This way, I can easily compare content across languages without getting lost.

And here's the kicker: ChatGPT is shaping up to be a solid translator, but no one’s integrated it into any frameworks yet. So, with my Django and Astrowind know-how (Astrowind is this cool UI built on Astro and Tailwind, and it’s open source), I whipped up a small Django library/framework to act as a headless CMS. I also tweaked the Astrowind source to pull in content from the CMS with multilingual support.

Check out this demo if you wanna see it in action:
https://django-astrowind.netlify.app/en

Not much has changed from the original Astrowind (https://astrowind.vercel.app/), except now you can switch languages on the fly, and the content is managed through a CMS with hashed caching.

If you’re interested in using it, I’ve put together a tutorial to help you set it up on your own machine:
https://django-headless-cms.readthedocs.io/en/latest/quick-start.html

Hope you dig it. And I hope my little project can be of some help to you guys.

Introducing Django Headless CMS: A Powerful and Flexible CMS Solution

Huy Nguyen — Thu, 20 Jun 2024 06:39:52 +0000

Introduction

Are you looking for an effortless way to create and manage a headless CMS? Look no further! I'm excited to introduce Django Headless CMS, a robust, open-source content management system designed to simplify the process of managing and delivering content across multiple platforms. Built on Django, this package leverages the sturdy foundation of the Django framework while providing a headless approach, making it ideal for modern web applications.

Why Choose Django Headless CMS?

Why choose Django-headless-cms over alternatives like Wagtail, Django-CMS, Strapi, or Contentful?

Headless CMS with minimal configuration: Unlike Wagtail and Django-CMS, which are primarily headed CMS solutions.
Responsive UI: Preferred over Strapi for its user interface.
Python & Django-based: Built on a robust framework with numerous extensions.
Open Source: Unlike Contentful and other paid services, it allows you to self-host your CMS.
Integration: Easily integrates with many existing Python and Django libraries.
Centralized multi-language support: Reduces redundancy and allows different content across multiple languages.

Key Features

Django Headless CMS comes packed with features to enhance your content management experience:

Schema Migrations: Manage content schema as database migrations, making it easier to sync from development to production environments.
Versioning Content: Revert to any previously saved version.
Publish/Draft Content: Manage published and draft content.
Markdown Editor Support: Enhanced content editing experience, useful for Posts/Articles.
Multi-language Support: Even for Markdown fields.
Auto Translate/Force Re-translate: Use ChatGPT or build your own translation interface.
Recursive Actions: Apply actions like Publish, Translate, and Force Re-translate to referenced objects.
Optimized Queries: Auto prefetch and select related queries for optimization.
Filter Published Objects: Easily filter to show only published content.
Auto Admin: Simplify admin page setup with features like:
- Sortable inline Many-to-Many (M2M) relationships.
- Sortable generic inlines.
- Display publish status of child objects.
Auto Serializer: Simplify serializer creation, including nested relations. Override specific serializers as needed.
Rapid API Development: Build APIs (Views/Viewsets) quickly and easily with auto serializers and optimized queries.
Easy Data Import/Export: Use the admin interface or management commands.
Auto-generated API Documentation & Playground: Automatically generate API documentation and an interactive playground.

Quick Start

To get started with Django Headless CMS, follow the tutorial from the docs:

Quick Start Guide

Demo

Want to see a quick demo? By the end of this tutorial, you will have a CMS that can be used to build a multi-lingual page that looks like this: Django Astrowind Demo

Some features:

Publish/Unpublish/Recursively publish:

Multi-language & markdown editor:

Drag-to-sort related items:
Auto-translate to multiple languages:

And other cool features. Read more in the docs.

Conclusion

After experimenting with various CMS frameworks, I built Django Headless CMS and Django Astrowind to meet my specific needs and address issues faced by others in the community.

For more information, explore the Django Headless CMS documentation and the Django Astrowind repository.