| Name | fastapi-distributed-websocket JSON |
| Version |
0.2.0
JSON |
| download |
| home_page | https://github.com/DontPanicO/fastapi-distributed-websocket |
| Summary | Large scale WebSocket with FastAPI |
| upload_time | 2023-02-07 09:20:50 |
| maintainer | |
| docs_url | None |
| author | Andrea Tedeschi |
| requires_python | >=3.10 |
| license | MIT License Copyright (c) 2022 Andrea Tedeschi Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
asyncio
fastapi
websocket
distributed
pubsub
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
redis
fastapi
websockets
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
[](https://www.python.org/)
[](https://mit-license.org/)
[](https://pypi.org/project/fastapi-distributed-websocket/)
# FastAPI Distributed Websocket
A library to implement websocket for distibuted systems based on FastAPI.
**N.B.: This library is still at an early stage, use it in production at your own risk.**
## What it does
The main features of this libarary are:
* Easly implementing broadcasting, pub/sub, chat rooms, etc...
* Proxy websocket connections to other servers (e.g. from an api gateway)
* Authentication
* Clean exception handling
* An in memory broker for fast development
## Problems of scaling websocket among multiple servers in production
Websocket is a relatively new protocol for real time communication over HTTP.
It establishes a durable, stateful, full-duplex connection between clients and the server.
It can be used to implement chats, real time notifications, broadcasting and
pub/sub models.
### Connections from clients
HTTP request/response mechanism fits very well to scale among multiple server
instances in production. Any time a client makes a request, it can connect
to any server instance and it's going to receive the same response. After
the response has been returned to the client, it went disconnected and it can
make another request without the need to hit the same instace as before.
This thanks to the stateless nature of HTTP.
However, Websocket establishes a stateful connection between the client and the
server and, if some error occurs and the connection went lost, we have to
ensure that clients are going to hit the same server instance they were connected
before, since that instance was managing the connection state.
**Stateful means that there is a state that can be manipulated. In particular,
a stateful connection is a connection that heavily relies on its state in
order to work**
### Broadcasting and group messages
Another problem of scaling Websocket occurs when we need to send messages to
multiple connected clients (i.e. broadcasting a message or sending a message to
all clients subscribed to a specific topic).
Imagine that we have a chat server, and that when an user send a message in a
specific chat room, we broadcast it to all users subscribed to that room.
If we have a single server instance, all connection are managed by this instance
so we can safely trust that the message will be delivered to all recipents.
On the other hand, with multiple server instances, users subscribing to a chat
room will probably connect to different instances. This way, if an user send a
message to the chat room *'xyz'* at the server *A*, users subscribed to the same
chat room at the server *B* are not receiving it.
### Documenting Websocket interfaces
Another common problem with Websocket, that's not even related to scaling, is
about documentation. Due to the event driven nature of the Websocket protocol
it does not fit well to be documented with [openapi](https://swagger.io/specification/).
However a new specification for asynchronous, event driven interfaces has been
defined recently. The spec name is [asyncapi](https://www.asyncapi.com/) and I'm
currently studying it. I don't know if this has to be implemented here or it's
better having a separate library for that, however this is surely something
we have to look at.
### Other problems
When I came first to think about this library, I started making a lot of research
of common problems related to Websocket on stackoverflow, reddit, github issues and
so on. I found some interesting resource that are however related to the implementation
itself. I picked up best solutions and elaborated my owns converging all of that in
this library.
## Examples
### Installation
`$ pip install fastapi-distributed-websocket`
### Basic usage
This is a basic example using an in memory broker with a single server instance.
```python
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, status
from distributed_websocket import Connection, WebSocketManager
app = FastAPI()
manager = WebSocketManager('channel:1', broker_url='memory://')
...
app.on_event('startup')
async def startup() -> None:
...
await manager.startup()
app.on_event('shutdown')
async def shutdown() -> None:
...
await manager.shutdown()
@app.websocket('/ws/{conn_id}')
async def websocket_endpoint(
ws: WebSocket,
conn_id: str,
*,
topic: Optional[Any] = None,
) -> None:
connection: Connection = await manager.new_connection(ws, conn_id)
try:
while True:
msg = await connection.receive_json()
await manager.broadcast(msg)
except WebSocketDisconnect:
await manager.remove_connection(connection)
```
The `manger.new_connection` method create a new Connection object and add it to
the `manager.active_connections` list. Note that after a `WebSocketDisconnect`
is raised, we call `remove_connection`: this method only remove the connection
object from the `manager.active_connections` list, without calling `connection.close`, since
the connection is already closed.
If you need to close a connection at any other time, you can use `manager.close_connection`.
If you use `connection.iter_json`, it already handles the `WebSocketDisconnect` exception, so
you can simply call `manager.remove_connection` just after the loop (see next code block).
Note that here we are using `manager.broadcast` to send the message to all connections managed
by the WebSocketManager instance. However, this method only work if we have a single server
instance. If we have multiple server instances, we have to use `manager.receive`, to properly
send the message to the broker.
```python
@app.websocket('/ws/{conn_id}')
async def websocket_endpoint(
ws: WebSocket,
conn_id: str,
*,
topic: Optional[Any] = None,
) -> None:
connection: Connection = await manager.new_connection(ws, conn_id)
# This is the preferred way of handling WebSocketDisconnect
async for msg in connection.iter_json():
await manager.receive(connection, msg)
await manager.remove_connection(connection)
```
### Proxy from an API gateway
Let's say we are developing a chat service and that all our services are behind
an API gateway. If we want to keep our websocket service behind it too, then
fastapi-distributed-websocket provides us with `WebSocketProxy`.
```python
from distributed_websocket import WebSocketProxy
# skipped other imports for brevity
app = FastAPI()
WS_TARGET_ENDPOINT = 'ws://websocket_service:8000/wshandler'
@app.websocket('/ws')
async def websocket_proxy(websocket: WebSocket):
await websocket.accept()
ws_proxy = WebSocketProxy(websocket, WS_TARGET_ENDPOINT)
await ws_proxy()
```
This will forward all messages from the client to the target endpoint and
all messages from the target endpoint to the client.
Now let's assume that our websocket service code is the code of our previous
example. Our API Gateway code will be:
```python
from distributed_websocket import WebSocketProxy
# skipped other imports for brevity
app = FastAPI()
WS_TARGET_ENDPOINT = 'ws://websocket_service:8000/ws/{}'
@app.websocket('/ws/{conn_id}')
async def websocket_endpoint(
ws: WebSocket,
conn_id: str,
) -> None:
await websocket.accept()
ws_proxy = WebSocketProxy(websocket, WS_TARGET_ENDPOINT.format(conn_id))
await ws_proxy()
```
## API Reference
### Connection
Connection objects wrap the websocket connection and provide a simple interface
to send and receive messages. They have a `topics` attribute to store subscriptions
patterns and implement pub/sub models.
* **`async`**` accept(self) -> None` \
Accept the connection.
* **`async`**` close(self, code: int = 1000) -> None` \
Close the connection with the specified status.
* **`async`**` receive_json(self) -> Any` \
Receive a JSON message.
* **`async`**` send_json(self, data: Any) -> None` \
Send a JSON message over the connection.
* **`async`**` iter_json(self) -> AsyncIterator[Any]` \
Iterate over the messages received over the connection.
### Messages
Message objects store the message type, the topic and the data. They provides
an easy serialization/deserialization mechanism.
Remeber that messages returned by `connection.iter_json` are already deserialized
into `dict` objects, so here we call *deserialization* the process of converting
a `dict` object into a `Message` object.
* `type: str` \
The message type.
* `topic: str` \
The message topic.
* `conn_id: str | list[str]` \
The connection id or list of connection ids that the message should be sent to.
* `data: Any` \
The message data.
* **`classmethod`**`from_client_message(cls, *, data: Any) -> Message` \
Create a message from a client message.
* `__serialize__(self) -> dict` \
Serialize the message into a `dict` object.
### Subscriptions
You can bind topics to connection objects to implement pub/sub models, notification and so on.
The `topics` attribute is a set of strings that follows the pattern matching syntax of MQTT.
This library share connection objects state between server instances, so you may find
references to terms like `channel`, `publish`, `subscribe` and `unsubscribe` referring to
the same concepts but applied to the underlying server/broker communication. \
This may be confusing, but remember to keep separated the communication between the server
and the clients, that you are developing and the communication between the server and the broker,
that you usually don't deal with.
* `subscribe(connection: Connection, message: Message) -> None` \
Subscribe a connection to `message.topic`.
* `unsubscribe(connection: Connection, message: Message) -> None` \
Unsubscribe a connection from `message.topic`.
* `hanlde_subscription_message(connection: Connection, message: Message) -> None` \
Calls `subscribe` or `unsubscribe` depending on the message type.
* `matches(topic: str, patterns: set[str]) -> bool` \
Check if `topic` matches any of the patterns in `patterns`.
### Authentication
Authentication is provided with the `WebSocketOAuth2PasswordBearer` class.
It inherits from *FastAPI* `OAuth2PasswordBearer` and overrides `__call__` method to accept
a `WebSocket` object.
* **`async`**` __call__(self, websocket: WebSocket) -> str | None` \
Authenticate the websocket connection and return the *Authorization* header value. \
If the authentication fails, return `None` if the objects has been initialized with `auto_error=False` \
or close the connection with the `WS_1008_POLICY_VIOLATION` code.
### Exceptions and Exception Handling
`fastapi-distributed-websocket` provides exception handling via decorators. You can use the
apposite decorators passing an exception class and a handler callable. Exception handlers
should accept only the exception object as argument.\
**Why this is useful?**\
Because sometimes the same type of exception can be raised by different parts of the application,
this way you can decorate the higer level function in the call stack to handle the exception at
any level.\
A base `WebSocketException` class is provided to bind connection objects to the exception, so
your handler function can easily access it.
If you need to access connection objects from the exception handler, your custom exceptions
should inherit from `WebSocketException`, no matter if they are really network related or not.
* `WebSocketException(self, message: str, *, connection: Connection) -> None`
* `InvalidSubscription(self, message: str, *, connection: Connection) -> None` \
Raised when a subscription pattern use an invalid syntax. Inherits from `WebSocketException`.
* `InvalidSubscriptionMessage(self, message: str, *, connection: Connection) -> None` \
Like `InvalidSubscription` it could be raised for bad syntax, but it could also be raised \
when the message type is not *subscribe* or *unsubscribe*. Inherits from `WebSocketException`.
* `handle(exc: BaseException, handler: Callable[..., Any]) -> Callable[..., Any]` \
Decorator to handle exceptions. If you decorate a function with this decorator, at any time \
an exception of type `exc` is raised or propagated to the function, it will be handled by `handler`. \
Use this decorator only if both your handler and the function are not async.
* **`async`**` ahandle(
exc: BaseException, handler: Callable[..., Coroutine[Any, Any, Any]]
) -> Callable[..., Any]` \
Decorator to handle exceptions, same ad `handle`, but the handler is a coroutine function. \
Use this if your handler is a coroutine function, while the decorated function could be \
either a sync or an async function.
### Broker Interfaces
Connections' state is shared between server instances using a pub/sub broker. By default,
the broker is a `reids.asyncio.Redis` instance (ex `aioredis.Redis`), but you can use any
other implementation. `fastapi-distributed-websocket` provides an `InMemoryBroker` class
for development purposes.
You can inherit from `BrokerInterface` and override the methods to implement your own broker.
* **`async`**` connect(self) -> Coroutine[Any, Any, None]` \
Connect to the broker.
* **`async`**` disconnect(self) -> Coroutine[Any, Any, None]` \
Disconnect from the broker.
* **`async`**` subscribe(self, channel: str) -> Coroutine[Any, Any, None]` \
Subscribe to a channel.
* **`async`**` unsubscribe(self, channel: str) -> Coroutine[Any, Any, None]` \
Unsubscribe from a channel.
* **`async`**` publish(self, channel: str, message: Any) -> Coroutine[Any, Any, None]` \
Publish a message to a channel.
* **`async`**` get_message(self, **kwargs) -> Coroutine[Any, Any, Message | None]` \
Get a message from the broker.
### WebSocketManager
The `WebSocketManager` class is where the main logic of the library is implemented. \
It keeps track of the connection objects and starts the broker connection.
It spawn a main task, a listener that wait (non-blocking) for messages from the broker,
and send them to the connection objects (broadcasting or checking for subscriptions)
spawning a new task for each send. \
The broker initialisation is done in the constructor while calls to `broker.connect` and
`broker.disconnect` are handled in the `startup` and `shutdown` methods.
* **`async`**` new_connection(
self, websocket: WebSocket, conn_id: str, topic: str | None = None
) -> Coroutine[Any, Any, Connection]` \
Create a new connection object, add it to `self.active_connections` and return it.
* **`async`**` close_connection(
self, connection: Connection, code: int = status.WS_1000_NORMAL_CLOSURE
) -> Coroutine[Any, Any, None]` \
Close a connection object and remove it from `self.active_connections`.
* ` remove_connection(self, connection: Connection) -> None` \
Remove a connection object from `self.active_connections`.
* `set_conn_id(self, connection: Connection, conn_id: str) -> None` \
Set the connection id and notify the client.
* `send(self, topic: str, message: Any) -> None` \
Send a message to all the connection objects subscribed to `topic`. \
It spawns a new task wrapping the coroutine resulting from `self._send`.
* `broadcast(self, message: Any) -> None` \
Send a message to all the connection objects. \
It spawns a new task wrapping the coroutine resulting from `self._broadcast`.
* `send_by_conn_id(self, conn_id: str | list[str], message: Any) -> None` \
Send a message to all the connection objects with the given connection id. \
It spawns a new task wrapping the coroutine resulting from `self._send_by_conn_id` \
if `conn_id` is a string or from `_send_multi_by_conn_id` if it is a list.
* `send_msg(self, message: Message) -> None` \
Based on the message type, it calls `send`, `send_by_conn_id` or `broadcast`.
* **`async`**` receive(
self, connection: Connection, message: Any
) -> Coroutine[Any, Any, None]` \
Receive a message from a connection object. It passes the message down to \
a private method that handle eventual subscriptions and then publish the message \
to the broker.
* **`async`**` startup(self) -> Coroutine[Any, Any, None]` \
Start the broker connection and the listener task.
* **`async`**` shutdown(self) -> Coroutine[Any, Any, None]` \
Close the broker connection and the listener task. \
It also takes care to cancel all the tasks spawned by `send` and `broadcast` and \
close all the connection objects before.
### WebSocketProxy
The `WebSocketProxy` class initialise callable objects that can be
used to start proxyng websocket messages from client to a server and viceversa.
It's initialised with a two parameters:
* **client**: a `WebSocket` object
* **server_endpoint**: a `str` containing the endpoint of the server
Notice that the target server could be a remote server or the same server that starts the proxy.
* **`async`**` __call__(self) -> Coroutine[Any, Any, None]` \
Start a websocket connection to **server_endpoint** and spawn two tasks: \
one that forwards the messages from the client to the target and the other that \
forwards the messages from the target to the client.
Raw data
{
"_id": null,
"home_page": "https://github.com/DontPanicO/fastapi-distributed-websocket",
"name": "fastapi-distributed-websocket",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": "",
"keywords": "asyncio,fastapi,websocket,distributed,pubsub",
"author": "Andrea Tedeschi",
"author_email": "Andrea Tedeschi <andrea.tedeschi@andreatedeschi.uno>",
"download_url": "https://files.pythonhosted.org/packages/31/92/118bd511272aaf331ba647cac318222bcabf5cdc507022bf9f6fe74cc3a3/fastapi-distributed-websocket-0.2.0.tar.gz",
"platform": "POSIX",
"description": "[](https://www.python.org/) \n[](https://mit-license.org/) \n[](https://pypi.org/project/fastapi-distributed-websocket/)\n\n# FastAPI Distributed Websocket\n\nA library to implement websocket for distibuted systems based on FastAPI.\n\n**N.B.: This library is still at an early stage, use it in production at your own risk.**\n\n\n## What it does\n\nThe main features of this libarary are:\n\n* Easly implementing broadcasting, pub/sub, chat rooms, etc...\n* Proxy websocket connections to other servers (e.g. from an api gateway)\n* Authentication\n* Clean exception handling\n* An in memory broker for fast development\n\n\n## Problems of scaling websocket among multiple servers in production\n\nWebsocket is a relatively new protocol for real time communication over HTTP.\nIt establishes a durable, stateful, full-duplex connection between clients and the server.\nIt can be used to implement chats, real time notifications, broadcasting and\npub/sub models.\n\n### Connections from clients\n\nHTTP request/response mechanism fits very well to scale among multiple server\ninstances in production. Any time a client makes a request, it can connect\nto any server instance and it's going to receive the same response. After\nthe response has been returned to the client, it went disconnected and it can\nmake another request without the need to hit the same instace as before.\nThis thanks to the stateless nature of HTTP.\n\nHowever, Websocket establishes a stateful connection between the client and the\nserver and, if some error occurs and the connection went lost, we have to\nensure that clients are going to hit the same server instance they were connected\nbefore, since that instance was managing the connection state.\n\n**Stateful means that there is a state that can be manipulated. In particular,\na stateful connection is a connection that heavily relies on its state in\norder to work**\n\n### Broadcasting and group messages\n\nAnother problem of scaling Websocket occurs when we need to send messages to\nmultiple connected clients (i.e. broadcasting a message or sending a message to\nall clients subscribed to a specific topic).\n\nImagine that we have a chat server, and that when an user send a message in a\nspecific chat room, we broadcast it to all users subscribed to that room.\nIf we have a single server instance, all connection are managed by this instance\nso we can safely trust that the message will be delivered to all recipents.\nOn the other hand, with multiple server instances, users subscribing to a chat\nroom will probably connect to different instances. This way, if an user send a\nmessage to the chat room *'xyz'* at the server *A*, users subscribed to the same\nchat room at the server *B* are not receiving it.\n\n### Documenting Websocket interfaces\n\nAnother common problem with Websocket, that's not even related to scaling, is\nabout documentation. Due to the event driven nature of the Websocket protocol\nit does not fit well to be documented with [openapi](https://swagger.io/specification/).\nHowever a new specification for asynchronous, event driven interfaces has been\ndefined recently. The spec name is [asyncapi](https://www.asyncapi.com/) and I'm\ncurrently studying it. I don't know if this has to be implemented here or it's\nbetter having a separate library for that, however this is surely something\nwe have to look at.\n\n### Other problems\n\nWhen I came first to think about this library, I started making a lot of research\nof common problems related to Websocket on stackoverflow, reddit, github issues and\nso on. I found some interesting resource that are however related to the implementation\nitself. I picked up best solutions and elaborated my owns converging all of that in\nthis library.\n\n## Examples\n\n### Installation\n\n`$ pip install fastapi-distributed-websocket`\n\n### Basic usage\n\nThis is a basic example using an in memory broker with a single server instance.\n\n```python\nfrom fastapi import FastAPI, WebSocket, WebSocketDisconnect, status\nfrom distributed_websocket import Connection, WebSocketManager\n\napp = FastAPI()\nmanager = WebSocketManager('channel:1', broker_url='memory://')\n...\n\n\napp.on_event('startup')\nasync def startup() -> None:\n ...\n await manager.startup()\n\n\napp.on_event('shutdown')\nasync def shutdown() -> None:\n ...\n await manager.shutdown()\n\n\n@app.websocket('/ws/{conn_id}')\nasync def websocket_endpoint(\n ws: WebSocket,\n conn_id: str,\n *,\n topic: Optional[Any] = None,\n) -> None:\n connection: Connection = await manager.new_connection(ws, conn_id)\n try:\n while True:\n msg = await connection.receive_json()\n await manager.broadcast(msg)\n except WebSocketDisconnect:\n await manager.remove_connection(connection)\n```\n\nThe `manger.new_connection` method create a new Connection object and add it to\nthe `manager.active_connections` list. Note that after a `WebSocketDisconnect`\nis raised, we call `remove_connection`: this method only remove the connection\nobject from the `manager.active_connections` list, without calling `connection.close`, since\nthe connection is already closed.\nIf you need to close a connection at any other time, you can use `manager.close_connection`.\nIf you use `connection.iter_json`, it already handles the `WebSocketDisconnect` exception, so\nyou can simply call `manager.remove_connection` just after the loop (see next code block).\n\nNote that here we are using `manager.broadcast` to send the message to all connections managed\nby the WebSocketManager instance. However, this method only work if we have a single server\ninstance. If we have multiple server instances, we have to use `manager.receive`, to properly\nsend the message to the broker.\n\n```python\n@app.websocket('/ws/{conn_id}')\nasync def websocket_endpoint(\n ws: WebSocket,\n conn_id: str,\n *,\n topic: Optional[Any] = None,\n) -> None:\n connection: Connection = await manager.new_connection(ws, conn_id)\n # This is the preferred way of handling WebSocketDisconnect\n async for msg in connection.iter_json():\n await manager.receive(connection, msg)\n await manager.remove_connection(connection)\n```\n\n### Proxy from an API gateway\n\nLet's say we are developing a chat service and that all our services are behind\nan API gateway. If we want to keep our websocket service behind it too, then\nfastapi-distributed-websocket provides us with `WebSocketProxy`.\n\n```python\nfrom distributed_websocket import WebSocketProxy\n# skipped other imports for brevity\n\napp = FastAPI()\n\n\nWS_TARGET_ENDPOINT = 'ws://websocket_service:8000/wshandler'\n\n@app.websocket('/ws')\nasync def websocket_proxy(websocket: WebSocket):\n await websocket.accept()\n ws_proxy = WebSocketProxy(websocket, WS_TARGET_ENDPOINT)\n await ws_proxy()\n```\n\nThis will forward all messages from the client to the target endpoint and\nall messages from the target endpoint to the client.\n\nNow let's assume that our websocket service code is the code of our previous\nexample. Our API Gateway code will be:\n\n```python\nfrom distributed_websocket import WebSocketProxy\n# skipped other imports for brevity\n\napp = FastAPI()\n\n\nWS_TARGET_ENDPOINT = 'ws://websocket_service:8000/ws/{}'\n\n@app.websocket('/ws/{conn_id}')\nasync def websocket_endpoint(\n ws: WebSocket,\n conn_id: str,\n) -> None:\n await websocket.accept()\n ws_proxy = WebSocketProxy(websocket, WS_TARGET_ENDPOINT.format(conn_id))\n await ws_proxy()\n```\n\n## API Reference\n\n### Connection\n\nConnection objects wrap the websocket connection and provide a simple interface\nto send and receive messages. They have a `topics` attribute to store subscriptions\npatterns and implement pub/sub models.\n\n* **`async`**` accept(self) -> None` \\\n Accept the connection.\n* **`async`**` close(self, code: int = 1000) -> None` \\\n Close the connection with the specified status.\n* **`async`**` receive_json(self) -> Any` \\\n Receive a JSON message.\n* **`async`**` send_json(self, data: Any) -> None` \\\n Send a JSON message over the connection.\n* **`async`**` iter_json(self) -> AsyncIterator[Any]` \\\n Iterate over the messages received over the connection.\n\n\n### Messages\n\nMessage objects store the message type, the topic and the data. They provides\nan easy serialization/deserialization mechanism.\nRemeber that messages returned by `connection.iter_json` are already deserialized\ninto `dict` objects, so here we call *deserialization* the process of converting\na `dict` object into a `Message` object.\n\n* `type: str` \\\n The message type.\n* `topic: str` \\\n The message topic.\n* `conn_id: str | list[str]` \\\n The connection id or list of connection ids that the message should be sent to.\n* `data: Any` \\\n The message data.\n\n* **`classmethod`**`from_client_message(cls, *, data: Any) -> Message` \\\n Create a message from a client message.\n* `__serialize__(self) -> dict` \\\n Serialize the message into a `dict` object.\n\n\n### Subscriptions\n\nYou can bind topics to connection objects to implement pub/sub models, notification and so on.\nThe `topics` attribute is a set of strings that follows the pattern matching syntax of MQTT.\nThis library share connection objects state between server instances, so you may find\nreferences to terms like `channel`, `publish`, `subscribe` and `unsubscribe` referring to\nthe same concepts but applied to the underlying server/broker communication. \\\nThis may be confusing, but remember to keep separated the communication between the server\nand the clients, that you are developing and the communication between the server and the broker,\nthat you usually don't deal with.\n\n* `subscribe(connection: Connection, message: Message) -> None` \\\n Subscribe a connection to `message.topic`.\n* `unsubscribe(connection: Connection, message: Message) -> None` \\\n Unsubscribe a connection from `message.topic`.\n* `hanlde_subscription_message(connection: Connection, message: Message) -> None` \\\n Calls `subscribe` or `unsubscribe` depending on the message type.\n* `matches(topic: str, patterns: set[str]) -> bool` \\\n Check if `topic` matches any of the patterns in `patterns`.\n\n\n### Authentication\n\nAuthentication is provided with the `WebSocketOAuth2PasswordBearer` class.\nIt inherits from *FastAPI* `OAuth2PasswordBearer` and overrides `__call__` method to accept\na `WebSocket` object.\n\n* **`async`**` __call__(self, websocket: WebSocket) -> str | None` \\\n Authenticate the websocket connection and return the *Authorization* header value. \\\n If the authentication fails, return `None` if the objects has been initialized with `auto_error=False` \\\n or close the connection with the `WS_1008_POLICY_VIOLATION` code.\n\n\n### Exceptions and Exception Handling\n\n`fastapi-distributed-websocket` provides exception handling via decorators. You can use the\napposite decorators passing an exception class and a handler callable. Exception handlers\nshould accept only the exception object as argument.\\\n**Why this is useful?**\\\nBecause sometimes the same type of exception can be raised by different parts of the application,\nthis way you can decorate the higer level function in the call stack to handle the exception at\nany level.\\\nA base `WebSocketException` class is provided to bind connection objects to the exception, so\nyour handler function can easily access it.\nIf you need to access connection objects from the exception handler, your custom exceptions\nshould inherit from `WebSocketException`, no matter if they are really network related or not.\n\n* `WebSocketException(self, message: str, *, connection: Connection) -> None`\n* `InvalidSubscription(self, message: str, *, connection: Connection) -> None` \\\n Raised when a subscription pattern use an invalid syntax. Inherits from `WebSocketException`.\n* `InvalidSubscriptionMessage(self, message: str, *, connection: Connection) -> None` \\\n Like `InvalidSubscription` it could be raised for bad syntax, but it could also be raised \\\n when the message type is not *subscribe* or *unsubscribe*. Inherits from `WebSocketException`.\n\n* `handle(exc: BaseException, handler: Callable[..., Any]) -> Callable[..., Any]` \\\n Decorator to handle exceptions. If you decorate a function with this decorator, at any time \\\n an exception of type `exc` is raised or propagated to the function, it will be handled by `handler`. \\\n Use this decorator only if both your handler and the function are not async.\n* **`async`**` ahandle(\n exc: BaseException, handler: Callable[..., Coroutine[Any, Any, Any]]\n) -> Callable[..., Any]` \\\n Decorator to handle exceptions, same ad `handle`, but the handler is a coroutine function. \\\n Use this if your handler is a coroutine function, while the decorated function could be \\\n either a sync or an async function.\n\n\n### Broker Interfaces\n\nConnections' state is shared between server instances using a pub/sub broker. By default,\nthe broker is a `reids.asyncio.Redis` instance (ex `aioredis.Redis`), but you can use any\nother implementation. `fastapi-distributed-websocket` provides an `InMemoryBroker` class\nfor development purposes.\nYou can inherit from `BrokerInterface` and override the methods to implement your own broker.\n\n* **`async`**` connect(self) -> Coroutine[Any, Any, None]` \\\n Connect to the broker.\n* **`async`**` disconnect(self) -> Coroutine[Any, Any, None]` \\\n Disconnect from the broker.\n* **`async`**` subscribe(self, channel: str) -> Coroutine[Any, Any, None]` \\\n Subscribe to a channel.\n* **`async`**` unsubscribe(self, channel: str) -> Coroutine[Any, Any, None]` \\\n Unsubscribe from a channel.\n* **`async`**` publish(self, channel: str, message: Any) -> Coroutine[Any, Any, None]` \\\n Publish a message to a channel.\n* **`async`**` get_message(self, **kwargs) -> Coroutine[Any, Any, Message | None]` \\\n Get a message from the broker.\n\n### WebSocketManager\n\nThe `WebSocketManager` class is where the main logic of the library is implemented. \\\nIt keeps track of the connection objects and starts the broker connection.\nIt spawn a main task, a listener that wait (non-blocking) for messages from the broker,\nand send them to the connection objects (broadcasting or checking for subscriptions)\nspawning a new task for each send. \\\nThe broker initialisation is done in the constructor while calls to `broker.connect` and\n`broker.disconnect` are handled in the `startup` and `shutdown` methods.\n\n* **`async`**` new_connection(\n self, websocket: WebSocket, conn_id: str, topic: str | None = None\n ) -> Coroutine[Any, Any, Connection]` \\\n Create a new connection object, add it to `self.active_connections` and return it.\n* **`async`**` close_connection(\n self, connection: Connection, code: int = status.WS_1000_NORMAL_CLOSURE\n ) -> Coroutine[Any, Any, None]` \\\n Close a connection object and remove it from `self.active_connections`.\n* ` remove_connection(self, connection: Connection) -> None` \\\n Remove a connection object from `self.active_connections`.\n* `set_conn_id(self, connection: Connection, conn_id: str) -> None` \\\n Set the connection id and notify the client.\n* `send(self, topic: str, message: Any) -> None` \\\n Send a message to all the connection objects subscribed to `topic`. \\\n It spawns a new task wrapping the coroutine resulting from `self._send`.\n* `broadcast(self, message: Any) -> None` \\\n Send a message to all the connection objects. \\\n It spawns a new task wrapping the coroutine resulting from `self._broadcast`.\n* `send_by_conn_id(self, conn_id: str | list[str], message: Any) -> None` \\\n Send a message to all the connection objects with the given connection id. \\\n It spawns a new task wrapping the coroutine resulting from `self._send_by_conn_id` \\\n if `conn_id` is a string or from `_send_multi_by_conn_id` if it is a list.\n* `send_msg(self, message: Message) -> None` \\\n Based on the message type, it calls `send`, `send_by_conn_id` or `broadcast`.\n* **`async`**` receive(\n self, connection: Connection, message: Any\n ) -> Coroutine[Any, Any, None]` \\\n Receive a message from a connection object. It passes the message down to \\\n a private method that handle eventual subscriptions and then publish the message \\\n to the broker.\n* **`async`**` startup(self) -> Coroutine[Any, Any, None]` \\\n Start the broker connection and the listener task.\n* **`async`**` shutdown(self) -> Coroutine[Any, Any, None]` \\\n Close the broker connection and the listener task. \\\n It also takes care to cancel all the tasks spawned by `send` and `broadcast` and \\\n close all the connection objects before.\n\n\n### WebSocketProxy\n\nThe `WebSocketProxy` class initialise callable objects that can be\nused to start proxyng websocket messages from client to a server and viceversa.\nIt's initialised with a two parameters:\n\n* **client**: a `WebSocket` object\n* **server_endpoint**: a `str` containing the endpoint of the server\n\nNotice that the target server could be a remote server or the same server that starts the proxy.\n\n* **`async`**` __call__(self) -> Coroutine[Any, Any, None]` \\\n Start a websocket connection to **server_endpoint** and spawn two tasks: \\\n one that forwards the messages from the client to the target and the other that \\\n forwards the messages from the target to the client.\n",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2022 Andrea Tedeschi Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ",
"summary": "Large scale WebSocket with FastAPI",
"version": "0.2.0",
"split_keywords": [
"asyncio",
"fastapi",
"websocket",
"distributed",
"pubsub"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b562db21fb5fb52ababa0001cf64b3b35681ec3b2fc37bc681b0b778fc13235d",
"md5": "88fe7b8f46ea86cb319c486dfcc1f061",
"sha256": "3c7e95cacebc534e611c68b336d7fe026d310dcb68a2879d7b1581c1f2050915"
},
"downloads": -1,
"filename": "fastapi_distributed_websocket-0.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "88fe7b8f46ea86cb319c486dfcc1f061",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 19160,
"upload_time": "2023-02-07T09:20:49",
"upload_time_iso_8601": "2023-02-07T09:20:49.252402Z",
"url": "https://files.pythonhosted.org/packages/b5/62/db21fb5fb52ababa0001cf64b3b35681ec3b2fc37bc681b0b778fc13235d/fastapi_distributed_websocket-0.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3192118bd511272aaf331ba647cac318222bcabf5cdc507022bf9f6fe74cc3a3",
"md5": "c103cdcdf9fca980ec30d474320d4891",
"sha256": "ef8753faf58f881d1036a95a2bee4a3b70fa808a0f118ff7cf14496384f2206a"
},
"downloads": -1,
"filename": "fastapi-distributed-websocket-0.2.0.tar.gz",
"has_sig": false,
"md5_digest": "c103cdcdf9fca980ec30d474320d4891",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 22315,
"upload_time": "2023-02-07T09:20:50",
"upload_time_iso_8601": "2023-02-07T09:20:50.716535Z",
"url": "https://files.pythonhosted.org/packages/31/92/118bd511272aaf331ba647cac318222bcabf5cdc507022bf9f6fe74cc3a3/fastapi-distributed-websocket-0.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-02-07 09:20:50",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "DontPanicO",
"github_project": "fastapi-distributed-websocket",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "redis",
"specs": [
[
"==",
"4.3.1"
]
]
},
{
"name": "fastapi",
"specs": [
[
">=",
"0.75.2"
]
]
},
{
"name": "websockets",
"specs": [
[
"==",
"10.3"
]
]
}
],
"lcname": "fastapi-distributed-websocket"
}
