# ssec
[](https://github.com/astral-sh/rye)
[](https://github.com/astral-sh/ruff)
[](https://github.com/python/mypy)
## Description
Yet another library for server-sent events.
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)).
## 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)
## Installation
ssec is written in [Python](https://www.python.org) and tries to keep track
of the newest version available. Currently[^3], this is
[Python 3.12.3](https://www.python.org/downloads/release/python-3123/).
On some operating systems, this version is pre-installed, but on many it is
not. This guide will not go into details on the installation process, but
there are tons of instructions out there to guide you. A good starting point
is the [beginners guide](https://www.python.org/about/gettingstarted/).
[^3]: 06\. May 2024
## Installation (User)
..via [rye](https://github.com/astral-sh/rye):
```sh
rye add ssec
```
..via [pip](https://pypi.org):
```sh
pip install ssec
```
## Installation (Developer)
**1\. Clone this repository to a desired location on your maschine using `ssh`:**
```sh
git git@github.com:sharly-project/ssec.git
```
**2\. Change into the project directory:**
```sh
cd ssec
```
**3\. Sync:**
```sh
rye sync
```
**4\. Run an example:**
```sh
rye run sse_sync_example
```
\- or -
```sh
rye run sse_async_example
```
**5\. Start coding!**
## Miscellaneous
### 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 that the [developers edition](#installation-developer)
> of `ssec` is installed and the virtual environment is running.
The documentation is then accessible via `doc/build/index.html`.
### Set up Visual Studio Code for Development
To edit the code base with [Visual Studio Code](https://code.visualstudio.com),
install the following extensions:
| Name | URL |
|-------------------|--------------------------------------------------------------------------------------|
| Python | <https://marketplace.visualstudio.com/items?itemName=ms-python.python> |
| Mypy Type Checker | <https://marketplace.visualstudio.com/items?itemName=ms-python.mypy-type-checker> |
| Ruff | <https://marketplace.visualstudio.com/items?itemName=charliermarsh.ruff> |
| markdownlint | <https://marketplace.visualstudio.com/items?itemName=DavidAnson.vscode-markdownlint> |
| Even Better TOML | <https://marketplace.visualstudio.com/items?itemName=tamasfe.even-better-toml> |
Necessary settings are already included in the `.vscode` directory and should
be enabled per 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://docs.python.org/3/library/unittest.html).
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 events, Network, Synchronous, Asynchronous",
"author": null,
"author_email": "\"J. Baudisch\" <justin.baudisch@hsbi.de>",
"download_url": "https://files.pythonhosted.org/packages/e6/e5/54f942d44348568f1d827b8f31cea9010003e1fe38957f9da77311c5f64f/ssec-4.1.0.tar.gz",
"platform": null,
"description": "# ssec\n\n[](https://github.com/astral-sh/rye)\n[](https://github.com/astral-sh/ruff)\n[](https://github.com/python/mypy)\n\n## Description\n\nYet another library for server-sent events. \nThis library works with [httpx](https://github.com/encode/httpx) to support\nsynchronous as well as asynchronous workflows but is also usable with other\nhttp frameworks ([see below](#aiohttp)).\n\n## Example\n\n`sync`\n\n```python\nimport logging\nimport ssec\n\ndef main() -> None:\n logging.basicConfig(level=logging.INFO)\n for event in ssec.sse(\n \"https://stream.wikimedia.org/v2/stream/recentchange\"\n ):\n print(event)\n\nmain()\n```\n\n`async`\n\n```python\nimport asyncio\nimport logging\nimport ssec\n\nasync def main() -> None:\n logging.basicConfig(level=logging.INFO)\n async for event in ssec.sse_async(\n \"https://stream.wikimedia.org/v2/stream/recentchange\"\n ):\n print(event)\n\nasyncio.run(main())\n```\n\n## Note\n\nAlthough there are already some libraries on the subject\n([aiohttp-sse-client](https://github.com/rtfol/aiohttp-sse-client),\n[aiosseclient](https://github.com/ebraminio/aiosseclient)), these are\nunfortunately not entirely correct. In example, both mentioned libraries\nasynchronously iterate over the stream content via `async for line in response.content`[^1][^2].\nThis internally calls [aiohttp](https://docs.aiohttp.org/en/stable)'s\n[`readuntil`](https://docs.aiohttp.org/en/stable/streams.html#aiohttp.StreamReader.readuntil)\nmethod with the default seperator `\\n`, but the official specification says:\n\n> Lines must be separated by either a U+000D CARRIAGE RETURN U+000A LINE FEED\n (CRLF) character pair, a single U+000A LINE FEED (LF) character, or a\n single +000D CARRIAGE RETURN (CR) character.\n\nAnother point is the error handling, which is often not sufficient to analyze\nthe error or is entirely skipped.\n\n### aiohttp\n\nAlthough this library works with `httpx`, it is also possible to use it with\nother http frameworks like `aiohttp` as long as they provide a method to\niterate over a byte-stream. Unfortunately, it is not possible to handle\nreconnection then, so you will have to implement that by yourself. An example\ncould look like this:\n\n```python\nimport asyncio\nimport logging\n\nimport aiohttp\nimport ssec\n\nasync def main() -> None:\n logging.basicConfig(level=logging.INFO)\n\n chunk_size = 1024\n connect_attempt = 1\n max_connect_attempts = 5\n config = ssec.SSEConfig(reconnect_timeout=3)\n async with aiohttp.ClientSession() as session:\n while True:\n headers = {\n \"Accept\": \"text/event-stream\",\n \"Cache-Control\": \"no-store\",\n }\n if config.last_event_id:\n headers[\"Last-Event-ID\"] = config.last_event_id\n try:\n async with session.get(\n \"https://stream.wikimedia.org/v2/stream/recentchange\",\n ) as response:\n streamer = response.content.iter_chunked(chunk_size)\n async for event in ssec.stream_async(streamer, config=config):\n print(event)\n except aiohttp.ClientError:\n if connect_attempt >= max_connect_attempts:\n logging.exception(\"Failed to connect!\")\n raise\n\n waiting_period = config.reconnect_timeout\n\n message = (\n f\"Failed to connect. \"\n f\"Reconnect in {waiting_period} seconds \"\n f\"[attempt {connect_attempt}/{max_connect_attempts}].\"\n )\n logging.info(message)\n\n connect_attempt += 1\n await asyncio.sleep(waiting_period)\n\nasyncio.run(main())\n```\n\n[^1]: [Code Reference](https://github.com/rtfol/aiohttp-sse-client/blob/e311075ac8b9b75d8b09512f8638f1dd03e2ef2b/aiohttp_sse_client/client.py#L157) \n[^2]: [Code Reference](https://github.com/ebraminio/aiosseclient/blob/375d597bcc3a7bf871b65913b366d515b300dc93/aiosseclient.py#L131)\n\n## Installation\n\nssec is written in [Python](https://www.python.org) and tries to keep track\nof the newest version available. Currently[^3], this is\n[Python 3.12.3](https://www.python.org/downloads/release/python-3123/).\nOn some operating systems, this version is pre-installed, but on many it is\nnot. This guide will not go into details on the installation process, but\nthere are tons of instructions out there to guide you. A good starting point\nis the [beginners guide](https://www.python.org/about/gettingstarted/).\n\n[^3]: 06\\. May 2024\n\n## Installation (User)\n\n..via [rye](https://github.com/astral-sh/rye):\n\n```sh\nrye add ssec\n```\n\n..via [pip](https://pypi.org):\n\n```sh\npip install ssec\n```\n\n## Installation (Developer)\n\n**1\\. Clone this repository to a desired location on your maschine using `ssh`:**\n\n```sh\ngit git@github.com:sharly-project/ssec.git\n```\n\n**2\\. Change into the project directory:**\n\n```sh\ncd ssec\n```\n\n**3\\. Sync:**\n\n```sh\nrye sync\n```\n\n**4\\. Run an example:**\n\n```sh\nrye run sse_sync_example\n```\n\n\\- or -\n\n```sh\nrye run sse_async_example\n```\n\n**5\\. Start coding!**\n\n## Miscellaneous\n\n### Documentation\n\nBuild the documentation by running the following command in the root directory\nof the project:\n\n```sh\nsphinx-build -b html docs/src docs/build\n```\n\n> The command requires that the [developers edition](#installation-developer)\n> of `ssec` is installed and the virtual environment is running.\n\nThe documentation is then accessible via `doc/build/index.html`.\n\n### Set up Visual Studio Code for Development\n\nTo edit the code base with [Visual Studio Code](https://code.visualstudio.com),\ninstall the following extensions:\n\n| Name | URL |\n|-------------------|--------------------------------------------------------------------------------------|\n| Python | <https://marketplace.visualstudio.com/items?itemName=ms-python.python> |\n| Mypy Type Checker | <https://marketplace.visualstudio.com/items?itemName=ms-python.mypy-type-checker> |\n| Ruff | <https://marketplace.visualstudio.com/items?itemName=charliermarsh.ruff> |\n| markdownlint | <https://marketplace.visualstudio.com/items?itemName=DavidAnson.vscode-markdownlint> |\n| Even Better TOML | <https://marketplace.visualstudio.com/items?itemName=tamasfe.even-better-toml> |\n\nNecessary settings are already included in the `.vscode` directory and should\nbe enabled per default.\n\n## Contributing\n\nContributing to `ssec` is highly appreciated, but comes with some requirements:\n\n1. **Type Hints**\n\n Write modern python code using\n [type annotations](https://peps.python.org/pep-0484/)\n to enable static analysis and potential runtime type checking.\n\n2. **Documentation**\n\n Write quality documentation using\n [numpydoc](https://numpydoc.readthedocs.io/en/latest/format.html)\n docstring conventions.\n\n3. **Linting**\n\n Lint your code with [ruff](https://github.com/charliermarsh/ruff) and\n [mypy](http://mypy-lang.org).\n\n4. **Style**\n\n Format your code using [ruff](https://github.com/charliermarsh/ruff).\n\n5. **Testing**\n\n Write tests for your code using\n [pytest](https://docs.python.org/3/library/unittest.html).\n",
"bugtrack_url": null,
"license": null,
"summary": "Synchronous and asynchronous library for server-sent events.",
"version": "4.1.0",
"project_urls": {
"Repository": "https://github.com/sharly-project/ssec"
},
"split_keywords": [
"server-sent events",
" network",
" synchronous",
" asynchronous"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "78c6a25e94f6d84f66a34bd1e5c45a873bc3dd3e560b8b1b3a22717d6a480c1f",
"md5": "e77c4732c2ce9786bb4f7ae28f7007bf",
"sha256": "61790b4551afe66e09a67da962ba2269e3382afde6ea198b7792e4a17eb61d01"
},
"downloads": -1,
"filename": "ssec-4.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "e77c4732c2ce9786bb4f7ae28f7007bf",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 10563,
"upload_time": "2024-06-04T07:55:00",
"upload_time_iso_8601": "2024-06-04T07:55:00.826026Z",
"url": "https://files.pythonhosted.org/packages/78/c6/a25e94f6d84f66a34bd1e5c45a873bc3dd3e560b8b1b3a22717d6a480c1f/ssec-4.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e6e554f942d44348568f1d827b8f31cea9010003e1fe38957f9da77311c5f64f",
"md5": "d403bff8d7f8efd251b87a3102d016c7",
"sha256": "7f3745827cd6cdfe1776ac750c9b3e2cec3f394e8fe8bf15981666dfa9844015"
},
"downloads": -1,
"filename": "ssec-4.1.0.tar.gz",
"has_sig": false,
"md5_digest": "d403bff8d7f8efd251b87a3102d016c7",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 13097,
"upload_time": "2024-06-04T07:55:02",
"upload_time_iso_8601": "2024-06-04T07:55:02.662018Z",
"url": "https://files.pythonhosted.org/packages/e6/e5/54f942d44348568f1d827b8f31cea9010003e1fe38957f9da77311c5f64f/ssec-4.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-06-04 07:55:02",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "sharly-project",
"github_project": "ssec",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "ssec"
}
JSON
