# Flask Streamy
`flask-streamy` is a lightweight Flask package that provides an abstraction for managing Server-Sent Events (SSE) streams. It allows you to easily create, manage, and terminate event streams in your Flask applications.
## Installation
```bash
pip install flask-streamy
```
## Basic Usage
```python
from flask import Flask
from flask_streamy import StreamManager
app = Flask(__name__)
sse = StreamManager()
@app.route('/send_event')
def send_event():
sse.send_event("stream1", data="Hello, World!", event_name="greeting")
return "Event sent"
@app.route('/stream')
def events():
headers = {
"X-Accel-Buffering": "no", # Disable response buffering
"Cache-Control": "no-cache"
}
return sse.get_stream("stream1", headers=headers)
@app.route('/end_stream')
def end_stream():
sse.end_stream("stream1")
return "Stream ended"
if __name__ == "__main__":
app.run(debug=True)
```
## API
**StreamManager**
* get_stream(stream_id, event_name="message", headers=None, keep_alive_interval=30):
* Retrieves or creates a stream for the given ID.
* Allows setting custom headers and a keep-alive interval to maintain the connection.
* Automatically sends a Connection: keep-alive header and periodic keep-alive messages.
* send_event(stream_id, data, event_name=None):
* Sends an event to the specified stream.
* Automatically handles errors with logging and retry logic.
* end_stream(stream_id):
* Ends the stream for the given ID.
* Logs the termination of the stream.
**SSE**
* Logging:
* Logs key events like stream creation, message addition, and stream termination.
* Error Handling with Retries:
* Automatically retries sending events up to a configurable number of times before ending the stream.
* Keep-Alive Messages:
* Periodic keep-alive messages are sent to prevent connection timeouts during idle periods.
## Testing
Here's an updated section for the `README.md` to include instructions on how to test the `flask-streamy` package:
### Running Tests
Tests for `flask-streamy` are located in the `tests` directory. To run the tests, you'll need to have `unittest` available, which is included in the Python standard library.
You can run the tests by executing the following command in the root of the package:
```bash
python -m unittest discover tests
```
### Writing Tests
The `tests/test_flask_streamy.py` file contains example unit tests that check the core functionality of `flask-streamy`. Here’s a brief overview of what the tests cover:
- **Stream Creation and Retrieval**: Ensures that streams are correctly created and retrieved by `get_stream`.
- **Sending Events**: Validates that events are correctly added to the stream's message queue.
- **Ending Streams**: Confirms that streams can be properly terminated and are no longer accessible after being ended.
You can add your own tests to the `tests` directory to cover additional use cases or to test custom functionality you may add to the package.
Raw data
{
"_id": null,
"home_page": "https://github.com/snickdx/flask-streamy",
"name": "flask-streamy",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": null,
"author": "Nicholas Mendez",
"author_email": "snickdx@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/e2/ff/e7e2de2abfe505344c94b1a759175e4fc1c768dd0bb857ed70b99c13a311/flask_streamy-0.1.2.tar.gz",
"platform": null,
"description": "# Flask Streamy\n\n`flask-streamy` is a lightweight Flask package that provides an abstraction for managing Server-Sent Events (SSE) streams. It allows you to easily create, manage, and terminate event streams in your Flask applications.\n\n## Installation\n\n```bash\npip install flask-streamy\n```\n\n## Basic Usage\n```python\nfrom flask import Flask\nfrom flask_streamy import StreamManager\n\napp = Flask(__name__)\nsse = StreamManager()\n\n@app.route('/send_event')\ndef send_event():\n sse.send_event(\"stream1\", data=\"Hello, World!\", event_name=\"greeting\")\n return \"Event sent\"\n\n@app.route('/stream')\ndef events():\n headers = {\n \"X-Accel-Buffering\": \"no\", # Disable response buffering\n \"Cache-Control\": \"no-cache\"\n }\n return sse.get_stream(\"stream1\", headers=headers)\n\n@app.route('/end_stream')\ndef end_stream():\n sse.end_stream(\"stream1\")\n return \"Stream ended\"\n\nif __name__ == \"__main__\":\n app.run(debug=True)\n```\n\n## API\n\n**StreamManager**\n\n* get_stream(stream_id, event_name=\"message\", headers=None, keep_alive_interval=30):\n * Retrieves or creates a stream for the given ID.\n * Allows setting custom headers and a keep-alive interval to maintain the connection.\n * Automatically sends a Connection: keep-alive header and periodic keep-alive messages.\n* send_event(stream_id, data, event_name=None):\n * Sends an event to the specified stream.\n * Automatically handles errors with logging and retry logic.\n* end_stream(stream_id):\n * Ends the stream for the given ID.\n * Logs the termination of the stream.\n\n**SSE**\n* Logging:\n * Logs key events like stream creation, message addition, and stream termination.\n* Error Handling with Retries:\n * Automatically retries sending events up to a configurable number of times before ending the stream.\n* Keep-Alive Messages:\n * Periodic keep-alive messages are sent to prevent connection timeouts during idle periods.\n\n## Testing\n\nHere's an updated section for the `README.md` to include instructions on how to test the `flask-streamy` package:\n\n\n### Running Tests\n\nTests for `flask-streamy` are located in the `tests` directory. To run the tests, you'll need to have `unittest` available, which is included in the Python standard library.\n\nYou can run the tests by executing the following command in the root of the package:\n\n```bash\npython -m unittest discover tests\n```\n\n### Writing Tests\n\nThe `tests/test_flask_streamy.py` file contains example unit tests that check the core functionality of `flask-streamy`. Here\u2019s a brief overview of what the tests cover:\n\n- **Stream Creation and Retrieval**: Ensures that streams are correctly created and retrieved by `get_stream`.\n- **Sending Events**: Validates that events are correctly added to the stream's message queue.\n- **Ending Streams**: Confirms that streams can be properly terminated and are no longer accessible after being ended.\n\nYou can add your own tests to the `tests` directory to cover additional use cases or to test custom functionality you may add to the package.\n",
"bugtrack_url": null,
"license": null,
"summary": "A Flask package for managing SSE streams.",
"version": "0.1.2",
"project_urls": {
"Homepage": "https://github.com/snickdx/flask-streamy"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "dfcd1bab9e95dd28d378d04ccd8dbc9ead7e153b4935f219020d7322f8c294cc",
"md5": "61d6fd3196b15015d4251b7faee9d423",
"sha256": "5e5deafb66e2e7aa81dbd329531720ea118a44e7a21d64f882829e93824efb97"
},
"downloads": -1,
"filename": "flask_streamy-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "61d6fd3196b15015d4251b7faee9d423",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 6415,
"upload_time": "2024-08-26T01:36:50",
"upload_time_iso_8601": "2024-08-26T01:36:50.700527Z",
"url": "https://files.pythonhosted.org/packages/df/cd/1bab9e95dd28d378d04ccd8dbc9ead7e153b4935f219020d7322f8c294cc/flask_streamy-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e2ffe7e2de2abfe505344c94b1a759175e4fc1c768dd0bb857ed70b99c13a311",
"md5": "44ff38e045822cb85338ab91003b6642",
"sha256": "f6d38693482aa0d5aa60c6067b46aef931ff8701a1854f2cfbaaeebd4b76528a"
},
"downloads": -1,
"filename": "flask_streamy-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "44ff38e045822cb85338ab91003b6642",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 5288,
"upload_time": "2024-08-26T01:36:51",
"upload_time_iso_8601": "2024-08-26T01:36:51.800601Z",
"url": "https://files.pythonhosted.org/packages/e2/ff/e7e2de2abfe505344c94b1a759175e4fc1c768dd0bb857ed70b99c13a311/flask_streamy-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-08-26 01:36:51",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "snickdx",
"github_project": "flask-streamy",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "flask-streamy"
}
JSON
