ClickUp Operator

Metadata-Version: 2.1 Name: sse-starlette Version: 2.1.3 Summary: SSE plugin for Starlette Author-Email: sysid <sysid@gmx.de> License: BSD-3-Clause Classifier: Development Status :: 5 - Production/Stable Classifier: Environment :: Web Environment Classifier: Intended Audience :: Developers Classifier: License :: OSI Approved :: BSD License Classifier: Operating System :: OS Independent Classifier: Programming Language :: Python :: 3 Classifier: Programming Language :: Python :: 3.9 Classifier: Programming Language :: Python :: 3.10 Classifier: Programming Language :: Python :: 3.11 Classifier: Programming Language :: Python :: 3.12 Classifier: Topic :: Internet :: WWW/HTTP Project-URL: Source, https://github.com/sysid/sse-starlette Requires-Python: >=3.8 Requires-Dist: starlette Requires-Dist: uvicorn Requires-Dist: anyio Requires-Dist: fastapi; extra == "examples" Provides-Extra: examples Description-Content-Type: text/markdown # 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