| Name | sse-starlette JSON |
| Version |
2.1.3
JSON |
| download |
| home_page | None |
| Summary | SSE plugin for Starlette |
| upload_time | 2024-08-01 08:52:50 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.8 |
| license | BSD-3-Clause |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Server Sent Events for [Starlette](https://github.com/encode/starlette) and [FastAPI](https://fastapi.tiangolo.com/)
[](https://pepy.tech/project/sse-starlette)
[![PyPI Version][pypi-image]][pypi-url]
[![Build Status][build-image]][build-url]
[![Code Coverage][coverage-image]][coverage-url]
> Implements the [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events) specification.
Background: https://sysid.github.io/server-sent-events/
Installation:
```shell
pip install sse-starlette
```
Usage:
```python
import asyncio
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Route
from sse_starlette.sse import EventSourceResponse
async def numbers(minimum, maximum):
for i in range(minimum, maximum + 1):
await asyncio.sleep(0.9)
yield dict(data=i)
async def sse(request):
generator = numbers(1, 5)
return EventSourceResponse(generator)
routes = [
Route("/", endpoint=sse)
]
app = Starlette(debug=True, routes=routes)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000, log_level='info')
```
Output:
**Caveat:** SSE streaming does not work in combination with [GZipMiddleware](https://github.com/encode/starlette/issues/20#issuecomment-704106436).
Be aware that for proper server shutdown your application must stop all
running tasks (generators). Otherwise you might experience the following warnings
at shutdown: `Waiting for background tasks to complete. (CTRL+C to force quit)`.
Client disconnects need to be handled in your Request handler (see example.py):
```python
async def endless(req: Request):
async def event_publisher():
i = 0
try:
while True:
i += 1
yield dict(data=i)
await asyncio.sleep(0.2)
except asyncio.CancelledError as e:
_log.info(f"Disconnected from client (via refresh/close) {req.client}")
# Do any other cleanup, if any
raise e
return EventSourceResponse(event_publisher())
```
## Special use cases
### Customize Ping
By default, the server sends a ping every 15 seconds. You can customize this by:
1. setting the `ping` parameter
2. by changing the `ping` event to a comment event so that it is not visible to the client
```python
@router.get("")
async def handle():
generator = numbers(1, 100)
return EventSourceResponse(
generator,
headers={"Server": "nini"},
ping=5,
ping_message_factory=lambda: ServerSentEvent(**{"comment": "You can't see\r\nthis ping"}),
)
```
### SSE Send Timeout
To avoid 'hanging' connections in case HTTP connection from a certain client was kept open, but the client
stopped reading from the connection you can specifiy a send timeout (see
[#89](https://github.com/sysid/sse-starlette/issues/89)).
```python
EventSourceResponse(..., send_timeout=5) # terminate hanging send call after 5s
```
### Fan out Proxies
Fan out proxies usually rely on response being cacheable. To support that, you can set the value of `Cache-Control`.
For example:
```python
return EventSourceResponse(
generator(), headers={"Cache-Control": "public, max-age=29"}
)
```
### Error Handling
See example: `examples/error_handling.py`
### Sending Responses without Async Generators
Async generators can expose tricky error and cleanup behavior especially when they are interrupted.
[Background: Cleanup in async generators](https://vorpus.org/blog/some-thoughts-on-asynchronous-api-design-in-a-post-asyncawait-world/#cleanup-in-generators-and-async-generators).
Example [`no_async_generators.py`](https://github.com/sysid/sse-starlette/pull/56#issue-1704495339) shows an alternative implementation
that does not rely on async generators but instead uses memory channels (`examples/no_async_generators.py`).
## Development, Contributing
1. install pdm: `pip install pdm`
2. install dependencies using pipenv: `pdm install -d`.
3. To run tests:
### Makefile
- make sure your virtualenv is active
- check `Makefile` for available commands and development support, e.g. run the unit tests:
```shell
make test
make tox
```
For integration testing you can use the provided examples in `tests` and `examples`.
If you are using Postman, please see: https://github.com/sysid/sse-starlette/issues/47#issuecomment-1445953826
<!-- Badges -->
[pypi-image]: https://badge.fury.io/py/sse-starlette.svg
[pypi-url]: https://pypi.org/project/sse-starlette/
[build-image]: https://github.com/sysid/sse-starlette/actions/workflows/build.yml/badge.svg
[build-url]: https://github.com/sysid/sse-starlette/actions/workflows/build.yml
[coverage-image]: https://codecov.io/gh/sysid/sse-starlette/branch/master/graph/badge.svg
[coverage-url]: https://codecov.io/gh/sysid/sse-starlette
Raw data
{
"_id": null,
"home_page": null,
"name": "sse-starlette",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": null,
"author": null,
"author_email": "sysid <sysid@gmx.de>",
"download_url": "https://files.pythonhosted.org/packages/72/fc/56ab9f116b2133521f532fce8d03194cf04dcac25f583cf3d839be4c0496/sse_starlette-2.1.3.tar.gz",
"platform": null,
"description": "# Server Sent Events for [Starlette](https://github.com/encode/starlette) and [FastAPI](https://fastapi.tiangolo.com/)\n\n[](https://pepy.tech/project/sse-starlette)\n[![PyPI Version][pypi-image]][pypi-url]\n[![Build Status][build-image]][build-url]\n[![Code Coverage][coverage-image]][coverage-url]\n\n> Implements the [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events) specification.\n\nBackground: https://sysid.github.io/server-sent-events/\n\nInstallation:\n\n```shell\npip install sse-starlette\n```\n\nUsage:\n\n```python\nimport asyncio\nimport uvicorn\nfrom starlette.applications import Starlette\nfrom starlette.routing import Route\nfrom sse_starlette.sse import EventSourceResponse\n\nasync def numbers(minimum, maximum):\n for i in range(minimum, maximum + 1):\n await asyncio.sleep(0.9)\n yield dict(data=i)\n\nasync def sse(request):\n generator = numbers(1, 5)\n return EventSourceResponse(generator)\n\nroutes = [\n Route(\"/\", endpoint=sse)\n]\n\napp = Starlette(debug=True, routes=routes)\n\nif __name__ == \"__main__\":\n uvicorn.run(app, host=\"0.0.0.0\", port=8000, log_level='info')\n```\n\nOutput:\n\n\n**Caveat:** SSE streaming does not work in combination with [GZipMiddleware](https://github.com/encode/starlette/issues/20#issuecomment-704106436).\n\nBe aware that for proper server shutdown your application must stop all\nrunning tasks (generators). Otherwise you might experience the following warnings\nat shutdown: `Waiting for background tasks to complete. (CTRL+C to force quit)`.\n\nClient disconnects need to be handled in your Request handler (see example.py):\n```python\nasync def endless(req: Request):\n async def event_publisher():\n i = 0\n try:\n while True:\n i += 1\n yield dict(data=i)\n await asyncio.sleep(0.2)\n except asyncio.CancelledError as e:\n _log.info(f\"Disconnected from client (via refresh/close) {req.client}\")\n # Do any other cleanup, if any\n raise e\n return EventSourceResponse(event_publisher())\n```\n\n## Special use cases\n### Customize Ping\nBy default, the server sends a ping every 15 seconds. You can customize this by:\n1. setting the `ping` parameter\n2. by changing the `ping` event to a comment event so that it is not visible to the client\n```python\n@router.get(\"\")\nasync def handle():\n generator = numbers(1, 100)\n return EventSourceResponse(\n generator,\n headers={\"Server\": \"nini\"},\n ping=5,\n ping_message_factory=lambda: ServerSentEvent(**{\"comment\": \"You can't see\\r\\nthis ping\"}),\n )\n```\n### SSE Send Timeout\nTo avoid 'hanging' connections in case HTTP connection from a certain client was kept open, but the client\nstopped reading from the connection you can specifiy a send timeout (see\n[#89](https://github.com/sysid/sse-starlette/issues/89)).\n```python\nEventSourceResponse(..., send_timeout=5) # terminate hanging send call after 5s\n```\n\n### Fan out Proxies\nFan out proxies usually rely on response being cacheable. To support that, you can set the value of `Cache-Control`.\nFor example:\n```python\nreturn EventSourceResponse(\n generator(), headers={\"Cache-Control\": \"public, max-age=29\"}\n )\n```\n### Error Handling\nSee example: `examples/error_handling.py`\n\n### Sending Responses without Async Generators\nAsync generators can expose tricky error and cleanup behavior especially when they are interrupted.\n\n[Background: Cleanup in async generators](https://vorpus.org/blog/some-thoughts-on-asynchronous-api-design-in-a-post-asyncawait-world/#cleanup-in-generators-and-async-generators).\n\nExample [`no_async_generators.py`](https://github.com/sysid/sse-starlette/pull/56#issue-1704495339) shows an alternative implementation\nthat does not rely on async generators but instead uses memory channels (`examples/no_async_generators.py`).\n\n\n## Development, Contributing\n1. install pdm: `pip install pdm`\n2. install dependencies using pipenv: `pdm install -d`.\n3. To run tests:\n\n### Makefile\n- make sure your virtualenv is active\n- check `Makefile` for available commands and development support, e.g. run the unit tests:\n```shell\nmake test\nmake tox\n```\n\nFor integration testing you can use the provided examples in `tests` and `examples`.\n\nIf you are using Postman, please see: https://github.com/sysid/sse-starlette/issues/47#issuecomment-1445953826\n\n\n<!-- Badges -->\n\n[pypi-image]: https://badge.fury.io/py/sse-starlette.svg\n[pypi-url]: https://pypi.org/project/sse-starlette/\n[build-image]: https://github.com/sysid/sse-starlette/actions/workflows/build.yml/badge.svg\n[build-url]: https://github.com/sysid/sse-starlette/actions/workflows/build.yml\n[coverage-image]: https://codecov.io/gh/sysid/sse-starlette/branch/master/graph/badge.svg\n[coverage-url]: https://codecov.io/gh/sysid/sse-starlette\n",
"bugtrack_url": null,
"license": "BSD-3-Clause",
"summary": "SSE plugin for Starlette",
"version": "2.1.3",
"project_urls": {
"Source": "https://github.com/sysid/sse-starlette"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "52aa36b271bc4fa1d2796311ee7c7283a3a1c348bad426d37293609ca4300eef",
"md5": "98c91b2a06d24a27123f22d72916a6f0",
"sha256": "8ec846438b4665b9e8c560fcdea6bc8081a3abf7942faa95e5a744999d219772"
},
"downloads": -1,
"filename": "sse_starlette-2.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "98c91b2a06d24a27123f22d72916a6f0",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 9383,
"upload_time": "2024-08-01T08:52:48",
"upload_time_iso_8601": "2024-08-01T08:52:48.659065Z",
"url": "https://files.pythonhosted.org/packages/52/aa/36b271bc4fa1d2796311ee7c7283a3a1c348bad426d37293609ca4300eef/sse_starlette-2.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "72fc56ab9f116b2133521f532fce8d03194cf04dcac25f583cf3d839be4c0496",
"md5": "7ed3e07a7580f56d0683ea2c20be3e8b",
"sha256": "9cd27eb35319e1414e3d2558ee7414487f9529ce3b3cf9b21434fd110e017169"
},
"downloads": -1,
"filename": "sse_starlette-2.1.3.tar.gz",
"has_sig": false,
"md5_digest": "7ed3e07a7580f56d0683ea2c20be3e8b",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 19678,
"upload_time": "2024-08-01T08:52:50",
"upload_time_iso_8601": "2024-08-01T08:52:50.248882Z",
"url": "https://files.pythonhosted.org/packages/72/fc/56ab9f116b2133521f532fce8d03194cf04dcac25f583cf3d839be4c0496/sse_starlette-2.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-08-01 08:52:50",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "sysid",
"github_project": "sse-starlette",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "sse-starlette"
}
