# Django Channels Router (Backend, PyPI)
A lightweight Django + Channels router that lets your WebSocket consumers behave like HTTP endpoints.
Designed to pair seamlessly with [`@djanext/observable-socket`](https://www.npmjs.com/package/@djanext/observable-socket).
---
## Features
- 🧭 Route names → handler methods (`sayHello` → `on_say_hello`)
- ⚡ Both sync & async consumers
- 🧩 Optional `hydrate` / `dehydrate` functions
- 🔁 Built-in heartbeat support (`PING`/`PONG`)
- 📦 Typed results and HTTP-style status codes
---
## Requirements
- **Python** ≥ 3.11
- **Django** ≥ 4.2
- **Channels** ≥ 4.0
- **Redis** (or any supported channel layer)
---
## Installation
```bash
pip install django-channels-router
```
In your project:
```py
# routing.py
from django.urls import re_path
from django_channels_router import SocketRouterConsumer
websocket_urlpatterns = [
re_path(r"ws/app/$", SocketRouterConsumer.as_asgi()),
]
```
---
## Example Consumer
```py
from django_channels_router import SocketRouterConsumer, StatusCodes
class AppSocket(SocketRouterConsumer):
@SocketRouterConsumer.routes.setter
def routes(self, _):
self._routes = [
{"route": "sayHello"},
{"route": "getArticle",
"hydrate": lambda headers, p: load_article(p["id"]),
"dehydrate": lambda art: {"id": art.id, "title": art.title}
},
]
def on_say_hello(self, payload, headers):
name = (payload or {}).get("name", "World")
return {"status": StatusCodes.OK, "payload": f"Hello, {name}!"}
```
**Async Example**
```py
from django_channels_router import AsyncSocketRouterConsumer
class AppSocketAsync(AsyncSocketRouterConsumer):
@AsyncSocketRouterConsumer.routes.setter
def routes(self, _):
self._routes = [{"route": "sayHello"}]
async def on_say_hello(self, payload, headers):
return {"status": 200, "payload": {"msg": "Hello async!"}}
```
---
## Serializing and Deserializing of messages
**hydrate** and **dehydrate** functions helps to focus only on logic part as these functions will handle deserializing and serializing automatically.
To use them, first you need to implement the converters (loader, serializer, ...)
```py
def load_article(headers, payload) -> Article | None:
try:
article_id = headers.get('id')
if not article_id:
return None
return Article.objects.get(id=article_id) # assume id is of type string like uuid
except Article.DoesNotExist:
return None
def deserialize(headers, payload) -> Article | None:
try:
serializer = ArticleSerializer(data=payload)
serializer.is_valid(raise_exception=True)
return serializer.save()
except serializers.ValidationError:
return None
def serialize(article: Article) -> str:
serializer = ArticleSerializer(article)
return serializer.data
```
now these converters can be used as follows:
```python
class ArticleConsumer(SocketRouterConsumer):
routes = [
{route: get, hydrate: load_article, dehydrate: serialize},
{route: create, hydrate: deserialize, dehydrate: serialize}
]
def on_get(headers, payload):
return {
payload: payload,
# article gets fetched from DB in background using the `load_article` function
# dehydrate will automatically serilize the article by running `serialize` function
status: StatusCodes.OK if payload else StatusCodes.NOT_FOUND
}
def on_create(headers, payload):
if not payload: # deserialize function has returned None
return {
payload: "Unable to create article",
status: StatusCodes.BAD_REQUEST
}
return {
payload: payload,
status: StatusCodes.CREATED
}
```
#### Note:
**hydrate** function only applies if returned status code by user is 200 series. (199 < status code < 300)
---
## Status Codes
| Symbol | Value | Meaning |
|---------|--------|---------|
| `StatusCodes.OK` | 200 | Success |
| `StatusCodes.BAD_REQUEST` | 400 | Malformed message |
| `StatusCodes.NOT_FOUND` | 404 | Unknown route |
| `StatusCodes.INTERNAL_SERVER_ERROR` | 500 | Handler failure |
---
## Good Practices
- Always `accept()` connections in `connect()` / `await accept()`.
- Access `result.get("payload")` safely — avoid missing keys.
- Use `@classmethod` route definitions if you plan to reuse the router base.
- Keep payload JSON-serializable.
---
## Frontend Client
Pair with `@djanext/observable-socket`
→ Handles auto-reconnect, multiple sockets, and `sendAndWait()` with Promises.
---
## License
MIT
Raw data
{
"_id": null,
"home_page": null,
"name": "django-channels-router",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": "Alireza Tabatabaeian <alireza.tabatabaeian@gmail.com>",
"keywords": "async, channels, django, observable-socket, real-time, router, websocket",
"author": null,
"author_email": "Alireza Tabatabaeian <alireza.tabatabaeian@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/27/4d/43c69dff6f6fd2a3137b94f4a83f32a5cb79bced4f4ffe43c933f201836f/django_channels_router-1.0.3.tar.gz",
"platform": null,
"description": "# Django Channels Router (Backend, PyPI)\n\nA lightweight Django + Channels router that lets your WebSocket consumers behave like HTTP endpoints. \nDesigned to pair seamlessly with [`@djanext/observable-socket`](https://www.npmjs.com/package/@djanext/observable-socket).\n\n---\n\n## Features\n- \ud83e\udded Route names \u2192 handler methods (`sayHello` \u2192 `on_say_hello`)\n- \u26a1 Both sync & async consumers\n- \ud83e\udde9 Optional `hydrate` / `dehydrate` functions\n- \ud83d\udd01 Built-in heartbeat support (`PING`/`PONG`)\n- \ud83d\udce6 Typed results and HTTP-style status codes\n\n---\n\n## Requirements\n- **Python** \u2265 3.11 \n- **Django** \u2265 4.2 \n- **Channels** \u2265 4.0 \n- **Redis** (or any supported channel layer)\n\n---\n\n## Installation\n\n```bash\npip install django-channels-router\n```\n\nIn your project:\n```py\n# routing.py\nfrom django.urls import re_path\nfrom django_channels_router import SocketRouterConsumer\n\nwebsocket_urlpatterns = [\n re_path(r\"ws/app/$\", SocketRouterConsumer.as_asgi()),\n]\n```\n\n---\n\n## Example Consumer\n\n```py\nfrom django_channels_router import SocketRouterConsumer, StatusCodes\n\nclass AppSocket(SocketRouterConsumer):\n @SocketRouterConsumer.routes.setter\n def routes(self, _):\n self._routes = [\n {\"route\": \"sayHello\"},\n {\"route\": \"getArticle\",\n \"hydrate\": lambda headers, p: load_article(p[\"id\"]),\n \"dehydrate\": lambda art: {\"id\": art.id, \"title\": art.title}\n },\n ]\n\n def on_say_hello(self, payload, headers):\n name = (payload or {}).get(\"name\", \"World\")\n return {\"status\": StatusCodes.OK, \"payload\": f\"Hello, {name}!\"}\n```\n\n**Async Example**\n```py\nfrom django_channels_router import AsyncSocketRouterConsumer\n\nclass AppSocketAsync(AsyncSocketRouterConsumer):\n @AsyncSocketRouterConsumer.routes.setter\n def routes(self, _):\n self._routes = [{\"route\": \"sayHello\"}]\n\n async def on_say_hello(self, payload, headers):\n return {\"status\": 200, \"payload\": {\"msg\": \"Hello async!\"}}\n```\n\n---\n\n## Serializing and Deserializing of messages\n\n**hydrate** and **dehydrate** functions helps to focus only on logic part as these functions will handle deserializing and serializing automatically.\nTo use them, first you need to implement the converters (loader, serializer, ...)\n\n```py\ndef load_article(headers, payload) -> Article | None:\n try:\n article_id = headers.get('id')\n if not article_id:\n return None\n return Article.objects.get(id=article_id) # assume id is of type string like uuid\n except Article.DoesNotExist:\n return None\n\ndef deserialize(headers, payload) -> Article | None:\n try:\n serializer = ArticleSerializer(data=payload)\n serializer.is_valid(raise_exception=True)\n return serializer.save()\n except serializers.ValidationError:\n return None\n\ndef serialize(article: Article) -> str:\n serializer = ArticleSerializer(article)\n return serializer.data\n```\n\nnow these converters can be used as follows:\n\n```python\n class ArticleConsumer(SocketRouterConsumer):\n routes = [\n {route: get, hydrate: load_article, dehydrate: serialize},\n {route: create, hydrate: deserialize, dehydrate: serialize}\n ]\n \n def on_get(headers, payload):\n return {\n payload: payload, \n # article gets fetched from DB in background using the `load_article` function\n # dehydrate will automatically serilize the article by running `serialize` function\n status: StatusCodes.OK if payload else StatusCodes.NOT_FOUND \n }\n \n def on_create(headers, payload):\n if not payload: # deserialize function has returned None\n return {\n payload: \"Unable to create article\",\n status: StatusCodes.BAD_REQUEST\n }\n \n return {\n payload: payload,\n status: StatusCodes.CREATED\n }\n```\n\n#### Note:\n**hydrate** function only applies if returned status code by user is 200 series. (199 < status code < 300) \n\n---\n\n## Status Codes\n\n| Symbol | Value | Meaning |\n|---------|--------|---------|\n| `StatusCodes.OK` | 200 | Success |\n| `StatusCodes.BAD_REQUEST` | 400 | Malformed message |\n| `StatusCodes.NOT_FOUND` | 404 | Unknown route |\n| `StatusCodes.INTERNAL_SERVER_ERROR` | 500 | Handler failure |\n\n---\n\n## Good Practices\n- Always `accept()` connections in `connect()` / `await accept()`.\n- Access `result.get(\"payload\")` safely \u2014 avoid missing keys.\n- Use `@classmethod` route definitions if you plan to reuse the router base.\n- Keep payload JSON-serializable.\n\n---\n\n## Frontend Client\n\nPair with `@djanext/observable-socket` \n\u2192 Handles auto-reconnect, multiple sockets, and `sendAndWait()` with Promises.\n\n---\n\n## License\n\nMIT",
"bugtrack_url": null,
"license": null,
"summary": "HTTP-style WebSocket routing for Django Channels.",
"version": "1.0.3",
"project_urls": {
"Documentation": "https://github.com/Alireza-Tabatabaeian/django-channels-router",
"Homepage": "https://github.com/Alireza-Tabatabaeian/django-channels-router",
"Issues": "https://github.com/Alireza-Tabatabaeian/django-channels-router/issues",
"Source": "https://github.com/Alireza-Tabatabaeian/django-channels-router"
},
"split_keywords": [
"async",
" channels",
" django",
" observable-socket",
" real-time",
" router",
" websocket"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "f682df0d9af3fa3196d467a22fc4b2f81d2765b33402c38e6e66aeb8ef30337c",
"md5": "987d97890f55b3603b7a82d42f7b2b7d",
"sha256": "25f1a4655c1ddbd265d08dbbefdd305f9113b7010a7c9c2a542c6c117159f358"
},
"downloads": -1,
"filename": "django_channels_router-1.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "987d97890f55b3603b7a82d42f7b2b7d",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 12208,
"upload_time": "2025-10-23T08:47:09",
"upload_time_iso_8601": "2025-10-23T08:47:09.169905Z",
"url": "https://files.pythonhosted.org/packages/f6/82/df0d9af3fa3196d467a22fc4b2f81d2765b33402c38e6e66aeb8ef30337c/django_channels_router-1.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "274d43c69dff6f6fd2a3137b94f4a83f32a5cb79bced4f4ffe43c933f201836f",
"md5": "718589c8f868f47c3c6a657b59f59a7a",
"sha256": "23d7b456260480c231ebb03993cb60f0a92a6a87ace5cf5f9903474584cdfa3f"
},
"downloads": -1,
"filename": "django_channels_router-1.0.3.tar.gz",
"has_sig": false,
"md5_digest": "718589c8f868f47c3c6a657b59f59a7a",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 7916,
"upload_time": "2025-10-23T08:47:10",
"upload_time_iso_8601": "2025-10-23T08:47:10.629533Z",
"url": "https://files.pythonhosted.org/packages/27/4d/43c69dff6f6fd2a3137b94f4a83f32a5cb79bced4f4ffe43c933f201836f/django_channels_router-1.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-23 08:47:10",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Alireza-Tabatabaeian",
"github_project": "django-channels-router",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "django-channels-router"
}
JSON
