# FastAPI WebSocker router
[](https://pypi.org/project/fastapi-ws-router)
[](https://pypi.org/project/fastapi-ws-router)
-----
Small library that allows one to document WebSocket messages in FastAPI.
## Overview
This library allows you to define websocket event handlers in the similar way one would define regular api endpoints. We
take somewhat opinionated approach and assume that all your events will confront to some PyDantic models, and it will be
possible to discriminate between them (preferably, based on some field. We use `pydantic.TypeAdapter` for this).
Library will make sure to generate OpenAPI documentation for your WebSocket handlers in a form of regular HTTP POST
endpoints. Because OpenAPI doesn't have any specifications for WebSockets, we have to bend some rules and use regular
routes to document possible WebSocket messages. See "OpenAPI limitations" section for more details.
## What this library does:
1. Provides somewhat opinionated way to document WebSocket endpoints in FastAPI.
2. Takes care of routing websocket messages to the corresponding handlers in the FastAPI-native way
3. Allows one to natively use PyDantic models to define WebSocket message schemas
4. Allows one to (somewhat) natively use FastAPI dependency injection
## What this library does not:
1. It doesn't take care of WebSockets management
2. It doesn't provide any kind of WebSocket server or client management
3. It doesn't handle any communications for you
In other words, you still have to take care of all the WebSocket operations you would normally do.
## Usage
Installation as usual:
```bash
pip install fastapi-ws-router
```
Then you can use it in your FastAPI application:
```python
from typing import Literal, Union
from fastapi import FastAPI
from fastapi_ws_router import WSRouter
from pydantic import BaseModel
# Messages we are expecting to receive defined as PyDantic models
class ChatMessage(BaseModel):
action: Literal["message"]
message: str
class ChatActivity(BaseModel):
action: Literal["activity"]
activity: str
app = FastAPI()
# Router to handle WebSocket connection
router = WSRouter(discriminator="action") # Discriminator is optional
# Handlers for specific messages
@router.receive(ChatMessage, callbacks=Union[ChatMessage, ChatActivity])
async def on_chat_message(websocket, data: ChatMessage):
await websocket.send_text(f"Got message: {data.message}")
@router.receive(ChatActivity)
async def on_chat_activity(websocket, data: ChatActivity):
await websocket.send_text(f"Got activity: {data.activity}")
# Finally, include the router in your FastAPI app (this should be the last step)
app.include_router(router, prefix="/ws")
```
In the example we use `action` field as a discriminator, although the message structure is completely up to
you. `discriminator` property is optional, it will help PyDantic to perform some optimizations
## Documenting server-side events
In cases when the WebSocket communication is bidirectional or server is emitting events, it can be desired to inform the
client what messages to expect. This can be achieved by providing a model(s) to the `callbacks` parameter.
```python
class Event1(BaseModel):
...
class Event2(BaseModel):
...
class Event3(BaseModel):
...
router = WSRouter(callbacks=Union[Event1, Event2])
@router.receive(Event1, callbacks=Union[Event2, Event3])
async def on_event1(websocket, data: Event1):
...
```
⚠️ Notice that those callbacks are informational only and pose no effect or restriction on the actual communication.
Server doesn't have to comply with them at all. They are there only for the documentation.
Callbacks defined in the router will be shown in the entrypoint route. This is to indicate that "once connected, client
can expect to receive these messages"
Callbacks defined on the event handlers will be shown in the corresponding route. This is to indicate that "once this
event is received, client can expect to receive these messages".
There is no "predefined" place to put events that are emitted by the server without any user interactions. It's up to
you to decide where to put them. Router callbacks might be a good place for that.
## WebSockets limitation
### Event handlers
This is the only thing we are somewhat opinionated about: event handler will always accept a single message being a
PyDantic model built from the received ws message (one message - one model instance).
Notice, that this doesn't apply to the messages emitted by the server. The library helps document them based on PyDantic
models, but it doesn't interfere with the actual communication in any way.
Event handler should always have next signature: `async def handler(WebSocket, BaseModel)` (first argument is always
a `WebSocket` instance and the second one is a PyDantic model instance)
Not-async handlers are not supported.
### Dependency injection
Due to the nature of WebSockets, only the entrypoint route (defined by the `WSRouter` itself) is able to apply
dependency injection. In other words, it is not possible to use any dependencies or `Path/Query/Header/Body` parameters
in the event handlers.
There is a way to pass down the data from the entrypoint to the handlers using the underlying `websocket.scope` object.
Below is an example of how one can pass the path parameter to the event handler:
```python
async def path_depends(
websocket: WebSocket,
item: str = Path(...), # This is a regular FastAPI dependency, everything is possible here
):
websocket.scope["path_item"] = item
router = WSRouter(dependencies=[Depends(path)]) # Inject dependency in the router
app.include_router(router, prefix="/ws/{item}") # Attach router to a parametrized path
@router.receive(ChatMessage)
async def on_chat_message(websocket: WebSocket, data: ChatMessage):
path_item = websocket.scope["path_item"] # Fetch path parameter from the scope
...
```
### Subroutes
It is not possible to attach or include any subroutes in the WebSocket route. However, one can have multiple `WSRouter`
instances attached to different paths.
## OpenAPI limitations
Currently, OpenAPI doesn't have any specification for the WebSockets. In order to include WebSocket events in the
documentation we ~~abuse~~ reuse regular `POST` endpoints.
These endpoints will have "weird" path (router prefix + handler name) - this provides some better visibility in the
documentation. Such routes, when attempted to be accessed directly, say, through the Swagegr UI, will never be found, as
they are not a real routes. (In reality, they are, they just "tweaked" to never match any path given)
It is possible to override path of each handler by providing `path` parameter in the `receive` decorator. It will be
appended to the router prefix. This path can be anything - handler routes are guaranteed to never match and requested
path. This is only for documentation purpose.
```python
router = WSRouter()
app.include_router(router, prefix="/ws")
@router.receive(ChatMessage, path=": WS Chat message") # Result in `/ws: WS Chat message` path in the documentation
async def on_chat_message(websocket, data: ChatMessage):
...
```
You can disable custom path by setting `path=""`.
WebSockets don't have a notion of a "response" similar to the http protocol, thus, by default, there will be no response
body in the OpenAPI specification. This can be modified with the `callbacks` parameter
We also do not support any status codes or response headers.
## Connection handlers
Connection handlers are exposed as decorators similar to the event handlers.
### `on_connect`
Emitted when a new WebSocket connection is established. Typically, this is where you determine whether to allow new
client to connect.
```python
@router.on_connect
async def on_connect(websocket: WebSocket):
# One must call either accept or close on the websocket
await websocket.accept()
```
### `on_disconnect`
Emitted when a WebSocket connection is closed by the client.
```python
@router.on_disconnect
async def on_disconnect(websocket: WebSocket, message: WebSocketDisconnect):
del my_connected_clients[websocket] # I.e., remove the client from the list of connected clients
```
### `on_fallback`
Emitted when we are unable to cast message to any of the known PyDantic models or there is a violation of the WebSocket
protocol. Message will be `None` in case of protocol violation. You will receive the original error in the third
parameter of the handler. `message` will always be either a string or bytes (based on what protocol you define in
the `WSRouter`)
In case of validation error, you will receive original PyDantic `ValidationError` as a third parameter.
```python
@router.on_fallback
async def on_fallback(websocket: WebSocket, message: Optional[Union[str, bytes]], err: Optional[Exception]):
...
```
## Dispatcher
It is possible to override the default dispatching behaviour. This might be needed in cases when you have a more
complicated handler selection logic.
`mapping` is a dict that contains all registered models mapping to the corresponding handlers. `message` is a raw
message received from the client (always `str` or `bytes`)
As the outcome, dispatcher most likely will call one of the handlers with the `websocket` and the deserialized message.
```python
# As we now use custom dispatcher, we can ignore the model assumption and use whatever we want in the arguments
# Be aware that this handler will still be inspected by FastAPI in order to build a documentation, so make sure that the arguments are "pydantic-compatible"
async def left_handler(websocket: WebSocket, message: str):
print("Left", message)
async def right_handler(websocket: WebSocket, message: str):
print("Right", message)
async def dispatcher(websocket: WebSocket, mapping: dict, message: str):
if message.startswith("LEFT-"):
await left_handler(websocket, message[5:])
else:
await right_handler(websocket, message[6:])
router = WSRouter(dispatcher=dispatcher)
app.include_router(router, prefix="/ws")
```
## Binary mode
By default, router assumes that messages are strings and use `websocket.receive_text()`.
It is possible to switch to bytes mode by providing `as_text=False` to the `WSRouter` constructor.
In this case `websocket.receive_bytes()` will be used instead.
In default dispatcher, received bytes will be sent to the PyDantic `TypeAdapter.validate_json` method.
## License
`fastapi-ws-router` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license.
Raw data
{
"_id": null,
"home_page": null,
"name": "fastapi-ws-router",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "documentation, fastapi, openapi, swagger, websockets, ws",
"author": null,
"author_email": "Jack Zabolotnyi <git@mclate.com>",
"download_url": "https://files.pythonhosted.org/packages/ee/ae/30f993ae6998f42c25416d3ba778f38b554633e2b74a79048e06179c82df/fastapi_ws_router-0.1.5.tar.gz",
"platform": null,
"description": "# FastAPI WebSocker router\n\n[](https://pypi.org/project/fastapi-ws-router)\n[](https://pypi.org/project/fastapi-ws-router)\n\n-----\n\nSmall library that allows one to document WebSocket messages in FastAPI.\n\n## Overview\n\nThis library allows you to define websocket event handlers in the similar way one would define regular api endpoints. We\ntake somewhat opinionated approach and assume that all your events will confront to some PyDantic models, and it will be\npossible to discriminate between them (preferably, based on some field. We use `pydantic.TypeAdapter` for this).\n\nLibrary will make sure to generate OpenAPI documentation for your WebSocket handlers in a form of regular HTTP POST\nendpoints. Because OpenAPI doesn't have any specifications for WebSockets, we have to bend some rules and use regular\nroutes to document possible WebSocket messages. See \"OpenAPI limitations\" section for more details.\n\n## What this library does:\n\n1. Provides somewhat opinionated way to document WebSocket endpoints in FastAPI.\n2. Takes care of routing websocket messages to the corresponding handlers in the FastAPI-native way\n3. Allows one to natively use PyDantic models to define WebSocket message schemas\n4. Allows one to (somewhat) natively use FastAPI dependency injection\n\n## What this library does not:\n\n1. It doesn't take care of WebSockets management\n2. It doesn't provide any kind of WebSocket server or client management\n3. It doesn't handle any communications for you\n\nIn other words, you still have to take care of all the WebSocket operations you would normally do.\n\n## Usage\n\nInstallation as usual:\n\n```bash\npip install fastapi-ws-router\n```\n\nThen you can use it in your FastAPI application:\n\n```python\nfrom typing import Literal, Union\n\nfrom fastapi import FastAPI\nfrom fastapi_ws_router import WSRouter\nfrom pydantic import BaseModel\n\n\n# Messages we are expecting to receive defined as PyDantic models\nclass ChatMessage(BaseModel):\n action: Literal[\"message\"]\n message: str\n\n\nclass ChatActivity(BaseModel):\n action: Literal[\"activity\"]\n activity: str\n\n\napp = FastAPI()\n\n# Router to handle WebSocket connection\nrouter = WSRouter(discriminator=\"action\") # Discriminator is optional\n\n\n# Handlers for specific messages\n@router.receive(ChatMessage, callbacks=Union[ChatMessage, ChatActivity])\nasync def on_chat_message(websocket, data: ChatMessage):\n await websocket.send_text(f\"Got message: {data.message}\")\n\n\n@router.receive(ChatActivity)\nasync def on_chat_activity(websocket, data: ChatActivity):\n await websocket.send_text(f\"Got activity: {data.activity}\")\n\n\n# Finally, include the router in your FastAPI app (this should be the last step)\napp.include_router(router, prefix=\"/ws\")\n\n```\n\n\n\nIn the example we use `action` field as a discriminator, although the message structure is completely up to\nyou. `discriminator` property is optional, it will help PyDantic to perform some optimizations\n\n## Documenting server-side events\n\nIn cases when the WebSocket communication is bidirectional or server is emitting events, it can be desired to inform the\nclient what messages to expect. This can be achieved by providing a model(s) to the `callbacks` parameter.\n\n```python\n\nclass Event1(BaseModel):\n ...\n\n\nclass Event2(BaseModel):\n ...\n\n\nclass Event3(BaseModel):\n ...\n\n\nrouter = WSRouter(callbacks=Union[Event1, Event2])\n\n\n@router.receive(Event1, callbacks=Union[Event2, Event3])\nasync def on_event1(websocket, data: Event1):\n ...\n\n```\n\n\u26a0\ufe0f Notice that those callbacks are informational only and pose no effect or restriction on the actual communication.\nServer doesn't have to comply with them at all. They are there only for the documentation.\n\nCallbacks defined in the router will be shown in the entrypoint route. This is to indicate that \"once connected, client\ncan expect to receive these messages\"\n\nCallbacks defined on the event handlers will be shown in the corresponding route. This is to indicate that \"once this\nevent is received, client can expect to receive these messages\".\n\nThere is no \"predefined\" place to put events that are emitted by the server without any user interactions. It's up to\nyou to decide where to put them. Router callbacks might be a good place for that.\n\n## WebSockets limitation\n\n### Event handlers\n\nThis is the only thing we are somewhat opinionated about: event handler will always accept a single message being a\nPyDantic model built from the received ws message (one message - one model instance).\n\nNotice, that this doesn't apply to the messages emitted by the server. The library helps document them based on PyDantic\nmodels, but it doesn't interfere with the actual communication in any way.\n\nEvent handler should always have next signature: `async def handler(WebSocket, BaseModel)` (first argument is always\na `WebSocket` instance and the second one is a PyDantic model instance)\n\nNot-async handlers are not supported.\n\n### Dependency injection\n\nDue to the nature of WebSockets, only the entrypoint route (defined by the `WSRouter` itself) is able to apply\ndependency injection. In other words, it is not possible to use any dependencies or `Path/Query/Header/Body` parameters\nin the event handlers.\n\nThere is a way to pass down the data from the entrypoint to the handlers using the underlying `websocket.scope` object.\nBelow is an example of how one can pass the path parameter to the event handler:\n\n```python\nasync def path_depends(\n websocket: WebSocket,\n item: str = Path(...), # This is a regular FastAPI dependency, everything is possible here\n):\n websocket.scope[\"path_item\"] = item\n\n\nrouter = WSRouter(dependencies=[Depends(path)]) # Inject dependency in the router\napp.include_router(router, prefix=\"/ws/{item}\") # Attach router to a parametrized path\n\n\n@router.receive(ChatMessage)\nasync def on_chat_message(websocket: WebSocket, data: ChatMessage):\n path_item = websocket.scope[\"path_item\"] # Fetch path parameter from the scope\n ...\n```\n\n### Subroutes\n\nIt is not possible to attach or include any subroutes in the WebSocket route. However, one can have multiple `WSRouter`\ninstances attached to different paths.\n\n## OpenAPI limitations\n\nCurrently, OpenAPI doesn't have any specification for the WebSockets. In order to include WebSocket events in the\ndocumentation we ~~abuse~~ reuse regular `POST` endpoints.\n\nThese endpoints will have \"weird\" path (router prefix + handler name) - this provides some better visibility in the\ndocumentation. Such routes, when attempted to be accessed directly, say, through the Swagegr UI, will never be found, as\nthey are not a real routes. (In reality, they are, they just \"tweaked\" to never match any path given)\n\nIt is possible to override path of each handler by providing `path` parameter in the `receive` decorator. It will be\nappended to the router prefix. This path can be anything - handler routes are guaranteed to never match and requested\npath. This is only for documentation purpose.\n\n```python\nrouter = WSRouter()\napp.include_router(router, prefix=\"/ws\")\n\n\n@router.receive(ChatMessage, path=\": WS Chat message\") # Result in `/ws: WS Chat message` path in the documentation\nasync def on_chat_message(websocket, data: ChatMessage):\n ...\n```\n\nYou can disable custom path by setting `path=\"\"`.\n\nWebSockets don't have a notion of a \"response\" similar to the http protocol, thus, by default, there will be no response\nbody in the OpenAPI specification. This can be modified with the `callbacks` parameter\n\nWe also do not support any status codes or response headers.\n\n## Connection handlers\n\nConnection handlers are exposed as decorators similar to the event handlers.\n\n### `on_connect`\n\nEmitted when a new WebSocket connection is established. Typically, this is where you determine whether to allow new\nclient to connect.\n\n```python\n\n@router.on_connect\nasync def on_connect(websocket: WebSocket):\n # One must call either accept or close on the websocket\n await websocket.accept()\n```\n\n### `on_disconnect`\n\nEmitted when a WebSocket connection is closed by the client.\n\n```python\n\n@router.on_disconnect\nasync def on_disconnect(websocket: WebSocket, message: WebSocketDisconnect):\n del my_connected_clients[websocket] # I.e., remove the client from the list of connected clients\n```\n\n### `on_fallback`\n\nEmitted when we are unable to cast message to any of the known PyDantic models or there is a violation of the WebSocket\nprotocol. Message will be `None` in case of protocol violation. You will receive the original error in the third\nparameter of the handler. `message` will always be either a string or bytes (based on what protocol you define in\nthe `WSRouter`)\n\nIn case of validation error, you will receive original PyDantic `ValidationError` as a third parameter.\n\n```python\n\n@router.on_fallback\nasync def on_fallback(websocket: WebSocket, message: Optional[Union[str, bytes]], err: Optional[Exception]):\n ...\n\n```\n\n## Dispatcher\n\nIt is possible to override the default dispatching behaviour. This might be needed in cases when you have a more\ncomplicated handler selection logic.\n\n`mapping` is a dict that contains all registered models mapping to the corresponding handlers. `message` is a raw\nmessage received from the client (always `str` or `bytes`)\n\nAs the outcome, dispatcher most likely will call one of the handlers with the `websocket` and the deserialized message.\n\n```python\n\n# As we now use custom dispatcher, we can ignore the model assumption and use whatever we want in the arguments\n# Be aware that this handler will still be inspected by FastAPI in order to build a documentation, so make sure that the arguments are \"pydantic-compatible\"\nasync def left_handler(websocket: WebSocket, message: str):\n print(\"Left\", message)\n\n\nasync def right_handler(websocket: WebSocket, message: str):\n print(\"Right\", message)\n\n\nasync def dispatcher(websocket: WebSocket, mapping: dict, message: str):\n if message.startswith(\"LEFT-\"):\n await left_handler(websocket, message[5:])\n else:\n await right_handler(websocket, message[6:])\n\n\nrouter = WSRouter(dispatcher=dispatcher)\napp.include_router(router, prefix=\"/ws\")\n\n```\n\n## Binary mode\n\nBy default, router assumes that messages are strings and use `websocket.receive_text()`.\nIt is possible to switch to bytes mode by providing `as_text=False` to the `WSRouter` constructor.\nIn this case `websocket.receive_bytes()` will be used instead.\nIn default dispatcher, received bytes will be sent to the PyDantic `TypeAdapter.validate_json` method.\n\n## License\n\n`fastapi-ws-router` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license.\n",
"bugtrack_url": null,
"license": null,
"summary": "Include WebSocket messages in OpenAPI documentation generated by FastAPI",
"version": "0.1.5",
"project_urls": {
"Documentation": "https://github.com/mclate/fastapi-ws-router#readme",
"Issues": "https://github.com/mclate/fastapi-ws-router/issues",
"Source": "https://github.com/mclate/fastapi-ws-router"
},
"split_keywords": [
"documentation",
" fastapi",
" openapi",
" swagger",
" websockets",
" ws"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "e7f183139a95f46e4f8964ff188e0a6810996a723a99ab03007f0ebb2da4d187",
"md5": "0555a6379b4aba5c8f7592a70a971e6e",
"sha256": "c7b49c310336ccabbcf54b793f967d4c7e279a434b1087334244e68289d5a5a0"
},
"downloads": -1,
"filename": "fastapi_ws_router-0.1.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0555a6379b4aba5c8f7592a70a971e6e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 10373,
"upload_time": "2024-09-08T19:42:37",
"upload_time_iso_8601": "2024-09-08T19:42:37.331337Z",
"url": "https://files.pythonhosted.org/packages/e7/f1/83139a95f46e4f8964ff188e0a6810996a723a99ab03007f0ebb2da4d187/fastapi_ws_router-0.1.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "eeae30f993ae6998f42c25416d3ba778f38b554633e2b74a79048e06179c82df",
"md5": "4a5f13766ebc73bcade5e87e420b0b35",
"sha256": "ecba8de4c934d89bd568ea7adf4538b29d0296e1a507f96f92d9b2a53ff2c8e1"
},
"downloads": -1,
"filename": "fastapi_ws_router-0.1.5.tar.gz",
"has_sig": false,
"md5_digest": "4a5f13766ebc73bcade5e87e420b0b35",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 508521,
"upload_time": "2024-09-08T19:42:38",
"upload_time_iso_8601": "2024-09-08T19:42:38.796000Z",
"url": "https://files.pythonhosted.org/packages/ee/ae/30f993ae6998f42c25416d3ba778f38b554633e2b74a79048e06179c82df/fastapi_ws_router-0.1.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-08 19:42:38",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "mclate",
"github_project": "fastapi-ws-router#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "fastapi-ws-router"
}
JSON
