flask-streamy 0.1.2

A Flask package for managing SSE streams.

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

pip install flask-streamy

Basic Usage

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:

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.