| Name | ssec JSON |
| Version |
4.2.2
JSON |
| download |
| home_page | None |
| Summary | Synchronous and asynchronous library for Server Sent Event (SSE). |
| upload_time | 2025-03-20 14:40:50 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.12 |
| license | None |
| keywords |
server sent event (sse)
sse-client
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
<p align="center">
<img src="https://github.com/ReMi-HSBI/ssec/blob/main/docs/src/static/images/ssec_logo.svg?raw=true" alt="drawing" width="33%"/>
</p>
# ssec
[](https://docs.astral.sh/ruff)
[](https://mypy-lang.org)
[](https://gitmoji.dev)
## Description
Python package for synchronous and asynchronous streaming of
[Server-Sent Events (SSE)](https://html.spec.whatwg.org/multipage/server-sent-events.html).
This library works with [httpx](https://github.com/encode/httpx)
to support synchronous as well as asynchronous workflows but is also
usable with other http frameworks ([see below](#aiohttp)).
## Installation
```sh
pip install ssec
```
## Example
`sync`
```python
import logging
import ssec
def main() -> None:
logging.basicConfig(level=logging.INFO)
for event in ssec.sse(
"https://stream.wikimedia.org/v2/stream/recentchange"
):
print(event)
main()
```
`async`
```python
import asyncio
import logging
import ssec
async def main() -> None:
logging.basicConfig(level=logging.INFO)
async for event in ssec.sse_async(
"https://stream.wikimedia.org/v2/stream/recentchange"
):
print(event)
asyncio.run(main())
```
## Note
Although there are already some libraries on the subject
([aiohttp-sse-client](https://github.com/rtfol/aiohttp-sse-client),
[aiosseclient](https://github.com/ebraminio/aiosseclient)), these are
unfortunately not entirely correct. In example, both mentioned libraries
asynchronously iterate over the stream content via
`async for line in response.content`[^1][^2].
This internally calls [aiohttp](https://docs.aiohttp.org/en/stable)'s
[`readuntil`](https://docs.aiohttp.org/en/stable/streams.html#aiohttp.StreamReader.readuntil)
method with the default seperator `\n`, but the official specification
says:
> Lines must be separated by either a U+000D CARRIAGE RETURN U+000A LINE
> FEED (CRLF) character pair, a single U+000A LINE FEED (LF) character,
> or a single +000D CARRIAGE RETURN (CR) character.
Another point is the error handling, which is often not sufficient to analyze
the error or is entirely skipped.
### aiohttp
Although this library works with `httpx`, it is also possible to use it
with other http frameworks like `aiohttp` as long as they provide a
method to iterate over a byte-stream. Unfortunately, it is not possible
to handle reconnection then, so you will have to implement that by
yourself. An example could look like this:
```python
import asyncio
import logging
import aiohttp
import ssec
async def main() -> None:
logging.basicConfig(level=logging.INFO)
chunk_size = 1024
connect_attempt = 1
max_connect_attempts = 5
config = ssec.SSEConfig(reconnect_timeout=3)
async with aiohttp.ClientSession() as session:
while True:
headers = {
"Accept": "text/event-stream",
"Cache-Control": "no-store",
}
if config.last_event_id:
headers["Last-Event-ID"] = config.last_event_id
try:
async with session.get(
"https://stream.wikimedia.org/v2/stream/recentchange",
) as response:
streamer = response.content.iter_chunked(chunk_size)
async for event in ssec.stream_async(streamer, config=config):
print(event)
except aiohttp.ClientError:
if connect_attempt >= max_connect_attempts:
logging.exception("Failed to connect!")
raise
waiting_period = config.reconnect_timeout
message = (
f"Failed to connect. "
f"Reconnect in {waiting_period} seconds "
f"[attempt {connect_attempt}/{max_connect_attempts}]."
)
logging.info(message)
connect_attempt += 1
await asyncio.sleep(waiting_period)
asyncio.run(main())
```
[^1]: [Code Reference](https://github.com/rtfol/aiohttp-sse-client/blob/e311075ac8b9b75d8b09512f8638f1dd03e2ef2b/aiohttp_sse_client/client.py#L157)
[^2]: [Code Reference](https://github.com/ebraminio/aiosseclient/blob/375d597bcc3a7bf871b65913b366d515b300dc93/aiosseclient.py#L131)
## Miscellaneous
### Installation (Developer)
```sh
pip install ssec[dev]
```
### Documentation
Build the documentation by running the following command in the root
directory of the project:
```sh
sphinx-build -b html docs/src docs/build
```
> The command requires the [developers edition](#installation-developer)
> of `ssec`.
The documentation is then accessible via `docs/build/index.html`.
### Set up Visual Studio Code for Development
To edit the code base with [Visual Studio Code](https://code.visualstudio.com),
you should install the `@recommended` extensions. Other necessary
settings are already included in the `.vscode` directory and should be
enabled by default.
## Contributing
Contributing to `ssec` is highly appreciated, but comes with some
requirements:
1. **Type Hints**
Write modern python code using
[type annotations](https://peps.python.org/pep-0484/)
to enable static analysis and potential runtime type checking.
2. **Documentation**
Write quality documentation using
[numpydoc](https://numpydoc.readthedocs.io/en/latest/format.html)
docstring conventions.
3. **Linting**
Lint your code with [ruff](https://github.com/charliermarsh/ruff) and
[mypy](http://mypy-lang.org).
4. **Style**
Format your code using [ruff](https://github.com/charliermarsh/ruff).
5. **Testing**
Write tests for your code using [pytest](https://pypi.org/project/pytest).
Raw data
{
"_id": null,
"home_page": null,
"name": "ssec",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": "\"J. Baudisch\" <justin.baudisch@hsbi.de>",
"keywords": "Server Sent Event (SSE), SSE-Client",
"author": null,
"author_email": "\"J. Baudisch\" <justin.baudisch@hsbi.de>",
"download_url": "https://files.pythonhosted.org/packages/a8/a4/8d48a287f768952ab8a846a9cbebb61de920273ff073f1268f8d7466e5b5/ssec-4.2.2.tar.gz",
"platform": null,
"description": "<p align=\"center\">\r\n <img src=\"https://github.com/ReMi-HSBI/ssec/blob/main/docs/src/static/images/ssec_logo.svg?raw=true\" alt=\"drawing\" width=\"33%\"/>\r\n</p>\r\n\r\n# ssec\r\n\r\n[](https://docs.astral.sh/ruff)\r\n[](https://mypy-lang.org)\r\n[](https://gitmoji.dev)\r\n\r\n## Description\r\n\r\nPython package for synchronous and asynchronous streaming of\r\n[Server-Sent Events (SSE)](https://html.spec.whatwg.org/multipage/server-sent-events.html).\r\nThis library works with [httpx](https://github.com/encode/httpx)\r\nto support synchronous as well as asynchronous workflows but is also\r\nusable with other http frameworks ([see below](#aiohttp)).\r\n\r\n## Installation\r\n\r\n```sh\r\npip install ssec\r\n```\r\n\r\n## Example\r\n\r\n`sync`\r\n\r\n```python\r\nimport logging\r\nimport ssec\r\n\r\ndef main() -> None:\r\n logging.basicConfig(level=logging.INFO)\r\n for event in ssec.sse(\r\n \"https://stream.wikimedia.org/v2/stream/recentchange\"\r\n ):\r\n print(event)\r\n\r\nmain()\r\n```\r\n\r\n`async`\r\n\r\n```python\r\nimport asyncio\r\nimport logging\r\nimport ssec\r\n\r\nasync def main() -> None:\r\n logging.basicConfig(level=logging.INFO)\r\n async for event in ssec.sse_async(\r\n \"https://stream.wikimedia.org/v2/stream/recentchange\"\r\n ):\r\n print(event)\r\n\r\nasyncio.run(main())\r\n```\r\n\r\n## Note\r\n\r\nAlthough there are already some libraries on the subject\r\n([aiohttp-sse-client](https://github.com/rtfol/aiohttp-sse-client),\r\n[aiosseclient](https://github.com/ebraminio/aiosseclient)), these are\r\nunfortunately not entirely correct. In example, both mentioned libraries\r\nasynchronously iterate over the stream content via\r\n`async for line in response.content`[^1][^2].\r\nThis internally calls [aiohttp](https://docs.aiohttp.org/en/stable)'s\r\n[`readuntil`](https://docs.aiohttp.org/en/stable/streams.html#aiohttp.StreamReader.readuntil)\r\nmethod with the default seperator `\\n`, but the official specification\r\nsays:\r\n\r\n> Lines must be separated by either a U+000D CARRIAGE RETURN U+000A LINE\r\n> FEED (CRLF) character pair, a single U+000A LINE FEED (LF) character,\r\n> or a single +000D CARRIAGE RETURN (CR) character.\r\n\r\nAnother point is the error handling, which is often not sufficient to analyze\r\nthe error or is entirely skipped.\r\n\r\n### aiohttp\r\n\r\nAlthough this library works with `httpx`, it is also possible to use it\r\nwith other http frameworks like `aiohttp` as long as they provide a\r\nmethod to iterate over a byte-stream. Unfortunately, it is not possible\r\nto handle reconnection then, so you will have to implement that by\r\nyourself. An example could look like this:\r\n\r\n```python\r\nimport asyncio\r\nimport logging\r\n\r\nimport aiohttp\r\nimport ssec\r\n\r\nasync def main() -> None:\r\n logging.basicConfig(level=logging.INFO)\r\n\r\n chunk_size = 1024\r\n connect_attempt = 1\r\n max_connect_attempts = 5\r\n config = ssec.SSEConfig(reconnect_timeout=3)\r\n async with aiohttp.ClientSession() as session:\r\n while True:\r\n headers = {\r\n \"Accept\": \"text/event-stream\",\r\n \"Cache-Control\": \"no-store\",\r\n }\r\n if config.last_event_id:\r\n headers[\"Last-Event-ID\"] = config.last_event_id\r\n try:\r\n async with session.get(\r\n \"https://stream.wikimedia.org/v2/stream/recentchange\",\r\n ) as response:\r\n streamer = response.content.iter_chunked(chunk_size)\r\n async for event in ssec.stream_async(streamer, config=config):\r\n print(event)\r\n except aiohttp.ClientError:\r\n if connect_attempt >= max_connect_attempts:\r\n logging.exception(\"Failed to connect!\")\r\n raise\r\n\r\n waiting_period = config.reconnect_timeout\r\n\r\n message = (\r\n f\"Failed to connect. \"\r\n f\"Reconnect in {waiting_period} seconds \"\r\n f\"[attempt {connect_attempt}/{max_connect_attempts}].\"\r\n )\r\n logging.info(message)\r\n\r\n connect_attempt += 1\r\n await asyncio.sleep(waiting_period)\r\n\r\nasyncio.run(main())\r\n```\r\n\r\n[^1]: [Code Reference](https://github.com/rtfol/aiohttp-sse-client/blob/e311075ac8b9b75d8b09512f8638f1dd03e2ef2b/aiohttp_sse_client/client.py#L157) \r\n[^2]: [Code Reference](https://github.com/ebraminio/aiosseclient/blob/375d597bcc3a7bf871b65913b366d515b300dc93/aiosseclient.py#L131)\r\n\r\n## Miscellaneous\r\n\r\n### Installation (Developer)\r\n\r\n```sh\r\npip install ssec[dev]\r\n```\r\n\r\n### Documentation\r\n\r\nBuild the documentation by running the following command in the root\r\ndirectory of the project:\r\n\r\n```sh\r\nsphinx-build -b html docs/src docs/build\r\n```\r\n\r\n> The command requires the [developers edition](#installation-developer)\r\n> of `ssec`.\r\n\r\nThe documentation is then accessible via `docs/build/index.html`.\r\n\r\n### Set up Visual Studio Code for Development\r\n\r\nTo edit the code base with [Visual Studio Code](https://code.visualstudio.com),\r\nyou should install the `@recommended` extensions. Other necessary\r\nsettings are already included in the `.vscode` directory and should be\r\nenabled by default.\r\n\r\n## Contributing\r\n\r\nContributing to `ssec` is highly appreciated, but comes with some\r\nrequirements:\r\n\r\n1. **Type Hints**\r\n\r\n Write modern python code using\r\n [type annotations](https://peps.python.org/pep-0484/)\r\n to enable static analysis and potential runtime type checking.\r\n\r\n2. **Documentation**\r\n\r\n Write quality documentation using\r\n [numpydoc](https://numpydoc.readthedocs.io/en/latest/format.html)\r\n docstring conventions.\r\n\r\n3. **Linting**\r\n\r\n Lint your code with [ruff](https://github.com/charliermarsh/ruff) and\r\n [mypy](http://mypy-lang.org).\r\n\r\n4. **Style**\r\n\r\n Format your code using [ruff](https://github.com/charliermarsh/ruff).\r\n\r\n5. **Testing**\r\n\r\n Write tests for your code using [pytest](https://pypi.org/project/pytest).\r\n",
"bugtrack_url": null,
"license": null,
"summary": "Synchronous and asynchronous library for Server Sent Event (SSE).",
"version": "4.2.2",
"project_urls": {
"Repository": "https://github.com/ReMi-HSBI/ssec"
},
"split_keywords": [
"server sent event (sse)",
" sse-client"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "17631537b7c8e1fc445490c40bbc26bf56020eeb4394340adab264ad6be5d8c6",
"md5": "babc86850e5b76aa2d4b3faf33b4cfca",
"sha256": "369a185b5de23d98e25c76d717b3be9ec16acb03c6be80476ee592409b462714"
},
"downloads": -1,
"filename": "ssec-4.2.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "babc86850e5b76aa2d4b3faf33b4cfca",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 10531,
"upload_time": "2025-03-20T14:40:47",
"upload_time_iso_8601": "2025-03-20T14:40:47.739995Z",
"url": "https://files.pythonhosted.org/packages/17/63/1537b7c8e1fc445490c40bbc26bf56020eeb4394340adab264ad6be5d8c6/ssec-4.2.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "a8a48d48a287f768952ab8a846a9cbebb61de920273ff073f1268f8d7466e5b5",
"md5": "a36778da86a786e41630d12eb8806060",
"sha256": "19b0005216c33a51b85065fd5e807156c7f18b93eb6198ef0b5915365b1323e3"
},
"downloads": -1,
"filename": "ssec-4.2.2.tar.gz",
"has_sig": false,
"md5_digest": "a36778da86a786e41630d12eb8806060",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 11630,
"upload_time": "2025-03-20T14:40:50",
"upload_time_iso_8601": "2025-03-20T14:40:50.788705Z",
"url": "https://files.pythonhosted.org/packages/a8/a4/8d48a287f768952ab8a846a9cbebb61de920273ff073f1268f8d7466e5b5/ssec-4.2.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-03-20 14:40:50",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "ReMi-HSBI",
"github_project": "ssec",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "ssec"
}
