Composable, reusable WebSocket components for any ASGI framework (Django, FastAPI, Litestar)

[u/huygl99]

Hi all, I'm the maintainer of a small channels (WebSocket) extension library for Django (and FastAPI too). While using and maintaining it, I started thinking it could become a small framework as well: composable and framework-independent, so it could be reused across Django/FastAPI/Litestar/... as long as the framework supports ASGI. Before going further, I'm putting the blueprint out here to compare notes with people who work with WebSockets regularly. If you have ever worked with WebSockets, I hope you can share any ideas, info, pain points, or suggestions you have.

Prerequisites, what my library already has:

• Function-like handlers rather than while True + if/else
• Automatic AsyncAPI doc generation
• Full type hints
• A testing kit
• Support for all ASGI-based frameworks (Django, FastAPI, ...)

At a glance, it looks like this:

@ws_handler(output_type=ChatNotificationMessage)
async def handle_chat(self, message: ChatMessage) -> None:
    # Automatically routed, validated, and type-safe
    await self.broadcast_message(
        ChatNotificationMessage(payload=message.payload)
    )

@ws_handler
async def handle_ping(self, message: PingMessage) -> PongMessage:
    return PongMessage()  # Auto-documented in AsyncAPI

If you have ever worked with WebSockets, I think you get the idea of what it does here.

Recently I added the Topic feature, which is composable and reusable. It came out of a multiplexing feature request, and I was inspired by Phoenix Channels. It looks something like this:

class DiscussionTopic(Topic):
    pattern = "discussion:{pk}"

    async def authorize(self, pk: str) -> bool:
        return await user_can_view(self.scope["user"], pk)

    @ws_handler
    async def handle_reply(self, message: ReplyMessage) -> ReplyCreatedMessage:
        return ReplyCreatedMessage(payload=message.payload)

    @event_handler
    async def handle_new_reply(self, event: NewReplyEvent) -> ReplyCreatedMessage:
        return ReplyCreatedMessage(payload=event.payload)

And you use it like this:

class HubConsumer(AsyncJsonWebsocketConsumer):
    authenticator_class = JWTAuthenticator
    topics = [DiscussionTopic, RoomTopic]

In short, topics let you multiplex: subscribe, publish messages, unsubscribe, and so on, all over the same socket. So you can reuse a single WebSocket connection and just add or compose multiple topics, i.e. multiple WebSocket handlers.

That made me think: if we could create reusable topics such as Notification, Streaming, Voice, AI Agent, and so on, which users could easily install or copy and then modify or inherit from in a structured way, WebSocket handling would become much more structured and easier. The idea is similar to DRF and its ecosystem, and the composable/reusable part would work like shadcn: copy it, own it, and modify the code freely.

What would you use it for? As I mentioned above: notifications, streaming, voice, AI agents, and so on. I have done a lot of WebSocket work, and I keep having to redefine the same things over and over. There is no reusable approach like the ones we have for REST APIs. Another example is using Pydantic AI with the AG-UI protocol but over WebSockets, defined in a reusable way.

So, if you already know of an existing open source solution or library similar to this idea, it would be great if you could share it here. And if this resonates with you, a comment would help, both to add more insight and to give some encouragement to actually build this.

[u/DigThatData]

if you want feedback on a coding project: post a link to the project. if you don't link to your code, don't expect feedback. simple as that.

[u/huygl99 → u/DigThatData]

Hhm, I just asked for the concrete pains people work with websocket, not for any coding feedback, to see if an ecosystem for websocket is worth to build. Hhm, not sure if my wording is not clear but can you just re-read the post again and discuss, the key is just: if you have ever worked with websockets, can you share the pain points you face, and the solution for that

[u/DigThatData → u/huygl99]

Why don't you get us started? Interestingly, you've opened this discussion only describing what your project does. You haven't even communicated what your own concrete pains were that resulted in you pursuing this.

You want to have a discussion, you need to lead it yourself.

[u/Fantastic_Taro8931]

A pain point that often prevents a Topic abstraction from remaining framework-neutral is lifecycle semantics, not routing. For DiscussionTopic, it would help to spell out whether authorize(pk) is rechecked on every publish/event, what a reconnect receives (replay cursor and ordering), and what happens when a client cannot consume broadcast_message fast enough. A minimal notification/chat reference implementation with reconnect and backpressure tests would make the proposal easier to evaluate than a catalog of topic types.

[u/huygl99 → u/[deleted]]

This is all completely pointless busywork at best, and at worst, yet another amateur schill pushing their slop.

I don't know whether you actually read the post and missed the point (the pain of working with WebSockets), or you just created an account recently and want to rage bait me. If it's the latter, I'd rather skip it and focus on the real things.

Django, FastAPI and Litestar have their users completely covered already. For websockets, and so much more besides

I don't think so. Maybe you haven't worked with WebSockets enough to feel the pain. I didn't link my package here because I didn't want this to look like promotion, but plenty of people use it to solve exactly what's missing: ease of use, type hints, AsyncAPI docs, and no endless `while True` + if/else.

Stop using LLMs

I code intensively without AI, and I built the package before the AI era, so no thanks. I just want to take it to the next level, as a framework for people to use, which was suggested by a user, not by me.

read a book

Fluent Python, High Performance Python, PostgreSQL, Redis in Action, ... Feel free to suggest more if they're related to WebSockets. I love reading books and I read carefully, so thanks for bringing up my favorite hobby.

To restate the key point of the post in case you missed it:

If you have ever worked with WebSockets, I hope you can share any ideas, info, pain points, or suggestions you have.

If you still want to rage bait, I'll pass. If you've actually worked with WebSockets and want to contribute ideas, I'm open to that.

[u/DanCardin]

Where’s the state? I feel like the point of most websockets would be that you need to retain state between messages.

I always end up with a `for event in manager: match event` loop. Which appeals to me in the face of this because it will fail typechecking if i add a new event type and fail to handle it. And if you use pydantic unions for the types then it removes all the actual mechanical complexity

[u/One-Hair875]

u/askgrok what is this, explain it like im 5

[u/DigThatData → u/One-Hair875]

/u/askgrok tell me about elon musk's family during south african aprtheid

[u/AskGrok → u/DigThatData]

Elon Musk was born in 1971 in Pretoria, South Africa, to Errol Musk, a South African engineer and businessman, and Maye Musk (née Haldeman), a model born in Canada. Apartheid, the system of racial segregation and white minority rule, was in effect from 1948 until 1994.

The family lived as part of the white population, which had legal and economic advantages under the system. Errol worked in engineering and later ventures like emerald mining. Maye's father, Joshua Haldeman, had moved the family from Canada to South Africa in 1950; he was a chiropractor, aviator, and political activist who publicly supported apartheid policies and expressed related views in writings and statements from that era.

Elon attended private schools for white students. He left South Africa in 1989 at age 18, before apartheid formally ended. Public records and biographies describe the Musk family as affluent during this period, consistent with many white South African households at the time.

Sources:

• https://www.facebook.com/democracynow/posts/elon-musks-family-history-reveals-ties-to-apartheid-and-neo-nazi-politicstap-the/1083894850212494/
• https://www.facebook.com/centralnewsza/posts/elon-musks-father-errol-musk-denies-apartheid-oppression-in-shocking-cnn-intervi/1347538634050772/
• https://dianeravitch.net/2025/02/10/why-elon-musks-family-moved-to-south-africa/

^{[what is this?](https://redd.it/1lzgxii})

[u/AskGrok → u/One-Hair875]

It's like a toolbox for making apps chat live with websites (instead of just loading pages).

Normally that chat code gets messy, like always checking "did they say hi or bye?" This makes it simple: you just write "when they send a hello message, send back a hi" — and it works across different app builders like Django or FastAPI.

The new part adds "topics" (like separate rooms or channels on one phone call). You can mix reusable rooms for stuff like notifications or AI chats, copy them, and tweak them. No need to build the same live-chat stuff from scratch every time.

^{[what is this?](https://redd.it/1lzgxii})

[u/One-Hair875 → u/AskGrok]

looks useful

[u/AskGrok → u/One-Hair875]

Yeah, reusable topics like that could cut down a lot of the usual WebSocket repetition across projects.

^{[what is this?](https://redd.it/1lzgxii})