# cjm-fasthtml-sse
<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->
## Install
``` bash
pip install cjm_fasthtml_sse
```
## Project Structure
nbs/
├── core.ipynb # Core SSE broadcast management system for FastHTML applications. Provides connection pooling, message distribution, and lifecycle hooks without UI dependencies.
├── dispatcher.ipynb # Event routing system with namespace support, pattern matching, and middleware pipeline. Enables decoupled event handling with priority-based execution and wildcard routing.
├── helpers.ipynb # Utility functions and decorators for common SSE patterns in FastHTML. Includes the @sse_element
├── htmx.ipynb # HTMX-specific SSE integration helpers for FastHTML. Simplifies adding SSE attributes, creating SSE-enabled elements, and managing HTMX SSE connections.
├── monitoring.ipynb # Connection monitoring and debugging tools for SSE applications. Provides configurable status indicators, automatic reconnection, and visibility change handling.
└── updater.ipynb # Flexible element update system for building out-of-band (OOB) swap elements. Register handlers by event type and compose updates without coupling to specific UI components.
Total: 6 notebooks
## Module Dependencies
``` mermaid
graph LR
core[core<br/>Core SSEBroadcastManager]
dispatcher[dispatcher<br/>SSEEventDispatcher]
helpers[helpers<br/>UI helpers & utilities]
htmx[htmx<br/>HTMXSSEConnector]
monitoring[monitoring<br/>Connection monitoring & config]
updater[updater<br/>SSEElementUpdater]
monitoring --> htmx
```
*1 cross-module dependencies detected*
## CLI Reference
No CLI commands found in this project.
## Module Overview
Detailed documentation for each module in the project:
### Core SSEBroadcastManager (`core.ipynb`)
> Core SSE broadcast management system for FastHTML applications.
> Provides connection pooling, message distribution, and lifecycle hooks
> without UI dependencies.
#### Import
``` python
from cjm_fasthtml_sse.core import (
SSEBroadcastManager
)
```
#### Classes
``` python
class SSEBroadcastManager:
def __init__(self,
max_queue_size: int = 100, # Maximum number of messages per connection queue
history_size: int = 50, # Number of broadcast messages to keep in history
default_timeout: float = 0.1 # Default timeout in seconds for queue operations
)
"""
Manages SSE connections and broadcasting without UI dependencies.
This class provides a reusable abstraction for managing Server-Sent Events
connections and broadcasting messages to multiple clients.
"""
def __init__(self,
max_queue_size: int = 100, # Maximum number of messages per connection queue
history_size: int = 50, # Number of broadcast messages to keep in history
default_timeout: float = 0.1 # Default timeout in seconds for queue operations
)
"Initialize the SSE Broadcast Manager."
async def register_connection(
self,
queue: Optional[asyncio.Queue] = None # Optional pre-existing queue, creates new one if not provided
) -> asyncio.Queue: # The queue associated with this connection
"Register a new SSE connection."
async def unregister_connection(
self,
queue: asyncio.Queue # The queue to unregister
)
"Unregister an SSE connection."
async def broadcast(self,
event_type: str, # Type of event being broadcast
data: Dict[str, Any], # Data to broadcast
timeout: Optional[float] = None # Optional timeout override for this broadcast
) -> int: # Number of successfully notified connections
"Broadcast a message to all connected clients."
def on_connect(
self,
callback: Callable # Function to call when a new connection is registered
)
"Register a callback for new connections."
def on_disconnect(
self,
callback: Callable # Function to call when a connection is unregistered
)
"Register a callback for disconnections."
def on_broadcast(
self,
callback: Callable # Function to call before broadcasting (can modify messages)
)
"Register a callback for broadcasts (can modify messages)."
def connection_count(
self
) -> int: # Number of active connections
"Get the current number of active connections."
def get_history(
self,
limit: Optional[int] = None # Optional limit on number of messages to return
) -> list[Dict[str, Any]]: # List of historical broadcast messages
"Get broadcast history."
```
### SSEEventDispatcher (`dispatcher.ipynb`)
> Event routing system with namespace support, pattern matching, and
> middleware pipeline. Enables decoupled event handling with
> priority-based execution and wildcard routing.
#### Import
``` python
from cjm_fasthtml_sse.dispatcher import (
SSEEvent,
SSEEventDispatcher
)
```
#### Classes
``` python
@dataclass
class SSEEvent:
"Represents an SSE event with metadata."
type: str
data: Dict[str, Any]
namespace: Optional[str]
priority: int = 0
timestamp: Optional[str]
def full_type(self):
"""Get the full event type including namespace."""
if self.namespace
"Get the full event type including namespace."
```
``` python
class SSEEventDispatcher:
def __init__(self):
"""Initialize the SSE Event Dispatcher."""
self._handlers: Dict[str, List[tuple[int, Callable]]] = {}
"""
Decoupled event routing system with namespace support,
middleware, filtering, and priority-based handling.
"""
def __init__(self):
"""Initialize the SSE Event Dispatcher."""
self._handlers: Dict[str, List[tuple[int, Callable]]] = {}
"Initialize the SSE Event Dispatcher."
def register_namespace(
self,
namespace: str # Namespace name to register for event organization
)
"Register a namespace for event organization."
def on(
self,
event_pattern: str, # Event pattern (supports wildcards: *, **)
priority: int = 0 # Handler priority (higher runs first)
)
"Decorator to register an event handler with pattern matching."
def add_handler(
self,
event_pattern: str, # Event pattern (e.g., "job:*", "**:completed")
handler: Callable, # Handler function
priority: int = 0 # Handler priority
)
"Add an event handler with pattern matching support."
def add_middleware(
self,
middleware: Callable # Function that takes (event, next) and calls next(event)
)
"Add middleware that processes events before handlers."
def add_filter(
self,
filter_func: Callable[[SSEEvent], bool] # Function that returns True to process event
)
"Add a filter to control which events are processed."
def add_transformer(
self,
transformer: Callable[[SSEEvent], SSEEvent] # Function that transforms an event
)
"Add a transformer to modify events before processing."
async def dispatch(
self,
event: Union[SSEEvent, Dict[str, Any]] # Event to dispatch (SSEEvent or dict)
) -> List[Any]: # List of handler results
"Dispatch an event through the processing pipeline."
def clear_handlers(
self,
pattern: Optional[str] = None # Specific pattern to clear, or None for all
)
"Clear handlers for a specific pattern or all handlers."
```
### UI helpers & utilities (`helpers.ipynb`)
> Utility functions and decorators for common SSE patterns in FastHTML.
> Includes the @sse_element
#### Import
``` python
from cjm_fasthtml_sse.helpers import (
oob_swap,
oob_element,
sse_element,
oob_update,
cleanup_sse_on_unload,
get_htmx_idx,
insert_htmx_sse_ext
)
```
#### Functions
``` python
def oob_swap(
"Add OOB swap attributes to an element."
```
``` python
def oob_element(
element_id: str, # ID of the target element
content: Any, # Content to swap
swap_type: str = "innerHTML" # Type of swap
)
"Create a wrapper element for OOB swap."
```
``` python
def sse_element(endpoint: str,
events: Optional[Union[str, List[str]]] = None, # Event name(s) to listen for from SSE stream
auto_close: bool = True, # Whether to auto-close on completion
swap_type: str = "message" # How to swap content
)
"Decorator to add SSE capabilities to any element."
```
``` python
def oob_update(
element_id: str, # Target element ID
content: Any, # Content to swap
swap_type: str = "innerHTML" # Type of swap (innerHTML, outerHTML, etc.)
)
"Create an out-of-band update element."
```
``` python
def cleanup_sse_on_unload(
) -> FT: # FastHTML element (Script) for cleanup
"Add script to cleanup SSE connections on page unload."
```
``` python
def get_htmx_idx(
hdrs: List # List of header elements to search
) -> int: # Index of HTMX script or -1 if not found
"Find the index of HTMX script in headers list."
```
``` python
def insert_htmx_sse_ext(
hdrs: List # List of header elements to modify
)
"Add HTMX SSE extension after HTMX script"
```
### HTMXSSEConnector (`htmx.ipynb`)
> HTMX-specific SSE integration helpers for FastHTML. Simplifies adding
> SSE attributes, creating SSE-enabled elements, and managing HTMX SSE
> connections.
#### Import
``` python
from cjm_fasthtml_sse.htmx import (
HTMXSSEConnector
)
```
#### Classes
``` python
class HTMXSSEConnector:
"""
Provides helper functions for setting up HTMX SSE connections
without hardcoding specific implementations.
"""
def add_sse_attrs(element,
endpoint: str, # SSE endpoint URL to connect to
events: Optional[Union[str, List[str]]] = None,
swap_type: str = "message", # How to swap content (message, innerHTML, outerHTML, etc.)
auto_reconnect: bool = True)
"Add SSE connection attributes to an element.
Args:
element: The element to add SSE attributes to
endpoint: SSE endpoint URL
events: Optional event name(s) to listen for
swap_type: How to swap content (message, innerHTML, outerHTML, etc.)
auto_reconnect: Whether to auto-reconnect on disconnect
Returns:
The element with SSE attributes added"
def create_sse_element(element_type=Div,
endpoint: str = None, # SSE endpoint URL to connect to
element_id: str = None, # Optional ID for the element
events: Optional[Union[str, List[str]]] = None,
swap_type: str = "message", # How to swap content when messages are received
hidden: bool = False, # Whether to hide the element initially
**kwargs)
"Create an element with SSE connection configured.
Args:
element_type: Type of element to create (Div, Span, etc.)
endpoint: SSE endpoint URL
element_id: Optional element ID
events: Optional event name(s) to listen for
swap_type: How to swap content
hidden: Whether to hide the element
**kwargs: Additional attributes for the element
Returns:
Element configured for SSE connection"
def sse_progress_element(job_id: str,
endpoint_template: str = "/stream_job_progress?job_id={job_id}", # URL template for the SSE endpoint
element_id_template: str = "progress-span-{job_id}", # Template for generating element ID
initial_content=None)
"Create an SSE-enabled progress element.
Args:
job_id: Job identifier
endpoint_template: Template for SSE endpoint URL
element_id_template: Template for element ID
initial_content: Initial content to display
Returns:
SSE-configured element for progress updates"
def sse_status_element(job_id: str,
endpoint_template: str = "/stream_job_status?job_id={job_id}", # URL template for the SSE endpoint
element_id_template: str = "status-span-{job_id}", # Template for generating element ID
initial_content=None)
"Create an SSE-enabled status element.
Args:
job_id: Job identifier
endpoint_template: Template for SSE endpoint URL
element_id_template: Template for element ID
initial_content: Initial content to display
Returns:
SSE-configured element for status updates"
def create_sse_monitor_script(
config: Dict[str, Any] # Configuration dictionary for monitoring setup
)
"Create a monitoring script for SSE connections.
Args:
config: Configuration dictionary with keys:
- sse_element_id: ID of SSE element to monitor
- status_element_id: ID of status display element
- auto_reconnect: Whether to auto-reconnect
- debug: Whether to enable debug logging
- status_indicators: Dict of status HTML strings
Returns:
Script element with monitoring code"
```
### Connection monitoring & config (`monitoring.ipynb`)
> Connection monitoring and debugging tools for SSE applications.
> Provides configurable status indicators, automatic reconnection, and
> visibility change handling.
#### Import
``` python
from cjm_fasthtml_sse.monitoring import (
SSEMonitorConfig,
create_sse_monitor
)
```
#### Functions
``` python
def create_sse_monitor(
htmx_sse: HTMXSSEConnector,
config: SSEMonitorConfig # SSEMonitorConfig instance
)
"Create a connection monitor with the specified configuration."
```
#### Classes
``` python
@dataclass
class SSEMonitorConfig:
"Configuration for SSE connection monitoring."
sse_element_id: str = 'sse-connection'
status_element_id: str = 'connection-status'
auto_reconnect: bool = True
reconnect_delay: int = 3000
debug: bool = False
heartbeat_timeout: int = 30000
status_indicators: Optional[Dict[str, str]]
```
### SSEElementUpdater (`updater.ipynb`)
> Flexible element update system for building out-of-band (OOB) swap
> elements. Register handlers by event type and compose updates without
> coupling to specific UI components.
#### Import
``` python
from cjm_fasthtml_sse.updater import (
SSEElementUpdater
)
```
#### Classes
``` python
class SSEElementUpdater:
def __init__(self):
"""Initialize the SSE Element Updater."""
self._handlers: Dict[str, List[Callable]] = {}
"""
Builds OOB swap elements without hardcoding UI components.
This class provides a flexible system for registering and executing
element update handlers based on event types.
"""
def __init__(self):
"""Initialize the SSE Element Updater."""
self._handlers: Dict[str, List[Callable]] = {}
"Initialize the SSE Element Updater."
def register(
self,
event_type: str, # The event type to handle
priority: int = 0 # Handler priority (higher numbers run first)
): # Decorator function
"Decorator to register an update handler for a specific event type."
def register_handler(
self,
event_type: str, # The event type to handle
handler: Callable, # The handler function
priority: int = 0 # Handler priority (higher numbers run first)
)
"Register an update handler programmatically."
def set_default_handler(
self,
handler: Callable # The default handler function
)
"Set a default handler for unregistered event types."
def add_preprocessor(
self,
processor: Callable # Function that processes (event_type, data) and returns modified data
)
"Add a preprocessor that runs before handlers."
def add_postprocessor(
self,
processor: Callable # Function that processes elements list and returns modified elements
)
"Add a postprocessor that runs after handlers."
def create_elements(
self,
event_type: str, # The type of event
data: Dict[str, Any] # Event data
) -> List[Any]: # List of elements to be sent via SSE
"Create elements for a given event type and data."
def clear_handlers(
self,
event_type: Optional[str] = None # Optional specific event type to clear
)
"Clear handlers for a specific event type or all handlers."
def get_registered_events(
self
) -> List[str]: # List of event types with registered handlers
"Get list of registered event types."
```
Raw data
{
"_id": null,
"home_page": "https://github.com/cj-mills/cjm-fasthtml-sse",
"name": "cjm-fasthtml-sse",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "nbdev jupyter notebook python",
"author": "Christian J. Mills",
"author_email": "9126128+cj-mills@users.noreply.github.com",
"download_url": "https://files.pythonhosted.org/packages/b0/11/e263d74d1a5ab389f4c63bf14b3ddb8283d2aac6e84911802ab9b17e6162/cjm_fasthtml_sse-0.0.20.tar.gz",
"platform": null,
"description": "# cjm-fasthtml-sse\n\n\n<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->\n\n## Install\n\n``` bash\npip install cjm_fasthtml_sse\n```\n\n## Project Structure\n\n nbs/\n \u251c\u2500\u2500 core.ipynb # Core SSE broadcast management system for FastHTML applications. Provides connection pooling, message distribution, and lifecycle hooks without UI dependencies.\n \u251c\u2500\u2500 dispatcher.ipynb # Event routing system with namespace support, pattern matching, and middleware pipeline. Enables decoupled event handling with priority-based execution and wildcard routing.\n \u251c\u2500\u2500 helpers.ipynb # Utility functions and decorators for common SSE patterns in FastHTML. Includes the @sse_element\n \u251c\u2500\u2500 htmx.ipynb # HTMX-specific SSE integration helpers for FastHTML. Simplifies adding SSE attributes, creating SSE-enabled elements, and managing HTMX SSE connections.\n \u251c\u2500\u2500 monitoring.ipynb # Connection monitoring and debugging tools for SSE applications. Provides configurable status indicators, automatic reconnection, and visibility change handling.\n \u2514\u2500\u2500 updater.ipynb # Flexible element update system for building out-of-band (OOB) swap elements. Register handlers by event type and compose updates without coupling to specific UI components.\n\nTotal: 6 notebooks\n\n## Module Dependencies\n\n``` mermaid\ngraph LR\n core[core<br/>Core SSEBroadcastManager]\n dispatcher[dispatcher<br/>SSEEventDispatcher]\n helpers[helpers<br/>UI helpers & utilities]\n htmx[htmx<br/>HTMXSSEConnector]\n monitoring[monitoring<br/>Connection monitoring & config]\n updater[updater<br/>SSEElementUpdater]\n\n monitoring --> htmx\n```\n\n*1 cross-module dependencies detected*\n\n## CLI Reference\n\nNo CLI commands found in this project.\n\n## Module Overview\n\nDetailed documentation for each module in the project:\n\n### Core SSEBroadcastManager (`core.ipynb`)\n\n> Core SSE broadcast management system for FastHTML applications.\n> Provides connection pooling, message distribution, and lifecycle hooks\n> without UI dependencies.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_sse.core import (\n SSEBroadcastManager\n)\n```\n\n#### Classes\n\n``` python\nclass SSEBroadcastManager:\n def __init__(self, \n max_queue_size: int = 100, # Maximum number of messages per connection queue\n history_size: int = 50, # Number of broadcast messages to keep in history\n default_timeout: float = 0.1 # Default timeout in seconds for queue operations\n )\n \"\"\"\n Manages SSE connections and broadcasting without UI dependencies.\n \n This class provides a reusable abstraction for managing Server-Sent Events\n connections and broadcasting messages to multiple clients.\n \"\"\"\n \n def __init__(self,\n max_queue_size: int = 100, # Maximum number of messages per connection queue\n history_size: int = 50, # Number of broadcast messages to keep in history\n default_timeout: float = 0.1 # Default timeout in seconds for queue operations\n )\n \"Initialize the SSE Broadcast Manager.\"\n \n async def register_connection(\n self,\n queue: Optional[asyncio.Queue] = None # Optional pre-existing queue, creates new one if not provided\n ) -> asyncio.Queue: # The queue associated with this connection\n \"Register a new SSE connection.\"\n \n async def unregister_connection(\n self,\n queue: asyncio.Queue # The queue to unregister\n )\n \"Unregister an SSE connection.\"\n \n async def broadcast(self, \n event_type: str, # Type of event being broadcast\n data: Dict[str, Any], # Data to broadcast\n timeout: Optional[float] = None # Optional timeout override for this broadcast\n ) -> int: # Number of successfully notified connections\n \"Broadcast a message to all connected clients.\"\n \n def on_connect(\n self,\n callback: Callable # Function to call when a new connection is registered\n )\n \"Register a callback for new connections.\"\n \n def on_disconnect(\n self,\n callback: Callable # Function to call when a connection is unregistered\n )\n \"Register a callback for disconnections.\"\n \n def on_broadcast(\n self,\n callback: Callable # Function to call before broadcasting (can modify messages)\n )\n \"Register a callback for broadcasts (can modify messages).\"\n \n def connection_count(\n self\n ) -> int: # Number of active connections\n \"Get the current number of active connections.\"\n \n def get_history(\n self,\n limit: Optional[int] = None # Optional limit on number of messages to return\n ) -> list[Dict[str, Any]]: # List of historical broadcast messages\n \"Get broadcast history.\"\n```\n\n### SSEEventDispatcher (`dispatcher.ipynb`)\n\n> Event routing system with namespace support, pattern matching, and\n> middleware pipeline. Enables decoupled event handling with\n> priority-based execution and wildcard routing.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_sse.dispatcher import (\n SSEEvent,\n SSEEventDispatcher\n)\n```\n\n#### Classes\n\n``` python\n@dataclass\nclass SSEEvent:\n \"Represents an SSE event with metadata.\"\n \n type: str\n data: Dict[str, Any]\n namespace: Optional[str]\n priority: int = 0\n timestamp: Optional[str]\n \n def full_type(self):\n \"\"\"Get the full event type including namespace.\"\"\"\n if self.namespace\n \"Get the full event type including namespace.\"\n```\n\n``` python\nclass SSEEventDispatcher:\n def __init__(self):\n \"\"\"Initialize the SSE Event Dispatcher.\"\"\"\n self._handlers: Dict[str, List[tuple[int, Callable]]] = {}\n \"\"\"\n Decoupled event routing system with namespace support,\n middleware, filtering, and priority-based handling.\n \"\"\"\n \n def __init__(self):\n \"\"\"Initialize the SSE Event Dispatcher.\"\"\"\n self._handlers: Dict[str, List[tuple[int, Callable]]] = {}\n \"Initialize the SSE Event Dispatcher.\"\n \n def register_namespace(\n self,\n namespace: str # Namespace name to register for event organization\n )\n \"Register a namespace for event organization.\"\n \n def on(\n self,\n event_pattern: str, # Event pattern (supports wildcards: *, **)\n priority: int = 0 # Handler priority (higher runs first)\n )\n \"Decorator to register an event handler with pattern matching.\"\n \n def add_handler(\n self,\n event_pattern: str, # Event pattern (e.g., \"job:*\", \"**:completed\")\n handler: Callable, # Handler function\n priority: int = 0 # Handler priority\n )\n \"Add an event handler with pattern matching support.\"\n \n def add_middleware(\n self,\n middleware: Callable # Function that takes (event, next) and calls next(event)\n )\n \"Add middleware that processes events before handlers.\"\n \n def add_filter(\n self,\n filter_func: Callable[[SSEEvent], bool] # Function that returns True to process event\n )\n \"Add a filter to control which events are processed.\"\n \n def add_transformer(\n self,\n transformer: Callable[[SSEEvent], SSEEvent] # Function that transforms an event\n )\n \"Add a transformer to modify events before processing.\"\n \n async def dispatch(\n self,\n event: Union[SSEEvent, Dict[str, Any]] # Event to dispatch (SSEEvent or dict)\n ) -> List[Any]: # List of handler results\n \"Dispatch an event through the processing pipeline.\"\n \n def clear_handlers(\n self,\n pattern: Optional[str] = None # Specific pattern to clear, or None for all\n )\n \"Clear handlers for a specific pattern or all handlers.\"\n```\n\n### UI helpers & utilities (`helpers.ipynb`)\n\n> Utility functions and decorators for common SSE patterns in FastHTML.\n> Includes the @sse_element\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_sse.helpers import (\n oob_swap,\n oob_element,\n sse_element,\n oob_update,\n cleanup_sse_on_unload,\n get_htmx_idx,\n insert_htmx_sse_ext\n)\n```\n\n#### Functions\n\n``` python\ndef oob_swap(\n \"Add OOB swap attributes to an element.\"\n```\n\n``` python\ndef oob_element(\n element_id: str, # ID of the target element\n content: Any, # Content to swap\n swap_type: str = \"innerHTML\" # Type of swap\n)\n \"Create a wrapper element for OOB swap.\"\n```\n\n``` python\ndef sse_element(endpoint: str, \n events: Optional[Union[str, List[str]]] = None, # Event name(s) to listen for from SSE stream\n auto_close: bool = True, # Whether to auto-close on completion\n swap_type: str = \"message\" # How to swap content\n )\n \"Decorator to add SSE capabilities to any element.\"\n```\n\n``` python\ndef oob_update(\n element_id: str, # Target element ID\n content: Any, # Content to swap\n swap_type: str = \"innerHTML\" # Type of swap (innerHTML, outerHTML, etc.)\n)\n \"Create an out-of-band update element.\"\n```\n\n``` python\ndef cleanup_sse_on_unload(\n) -> FT: # FastHTML element (Script) for cleanup\n \"Add script to cleanup SSE connections on page unload.\"\n```\n\n``` python\ndef get_htmx_idx(\n hdrs: List # List of header elements to search\n) -> int: # Index of HTMX script or -1 if not found\n \"Find the index of HTMX script in headers list.\"\n```\n\n``` python\ndef insert_htmx_sse_ext(\n hdrs: List # List of header elements to modify\n)\n \"Add HTMX SSE extension after HTMX script\"\n```\n\n### HTMXSSEConnector (`htmx.ipynb`)\n\n> HTMX-specific SSE integration helpers for FastHTML. Simplifies adding\n> SSE attributes, creating SSE-enabled elements, and managing HTMX SSE\n> connections.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_sse.htmx import (\n HTMXSSEConnector\n)\n```\n\n#### Classes\n\n``` python\nclass HTMXSSEConnector:\n \"\"\"\n Provides helper functions for setting up HTMX SSE connections\n without hardcoding specific implementations.\n \"\"\"\n \n def add_sse_attrs(element,\n endpoint: str, # SSE endpoint URL to connect to\n events: Optional[Union[str, List[str]]] = None,\n swap_type: str = \"message\", # How to swap content (message, innerHTML, outerHTML, etc.)\n auto_reconnect: bool = True)\n \"Add SSE connection attributes to an element.\n\nArgs:\n element: The element to add SSE attributes to\n endpoint: SSE endpoint URL\n events: Optional event name(s) to listen for\n swap_type: How to swap content (message, innerHTML, outerHTML, etc.)\n auto_reconnect: Whether to auto-reconnect on disconnect\n \nReturns:\n The element with SSE attributes added\"\n \n def create_sse_element(element_type=Div,\n endpoint: str = None, # SSE endpoint URL to connect to\n element_id: str = None, # Optional ID for the element\n events: Optional[Union[str, List[str]]] = None,\n swap_type: str = \"message\", # How to swap content when messages are received\n hidden: bool = False, # Whether to hide the element initially\n **kwargs)\n \"Create an element with SSE connection configured.\n\nArgs:\n element_type: Type of element to create (Div, Span, etc.)\n endpoint: SSE endpoint URL\n element_id: Optional element ID\n events: Optional event name(s) to listen for\n swap_type: How to swap content\n hidden: Whether to hide the element\n **kwargs: Additional attributes for the element\n \nReturns:\n Element configured for SSE connection\"\n \n def sse_progress_element(job_id: str,\n endpoint_template: str = \"/stream_job_progress?job_id={job_id}\", # URL template for the SSE endpoint\n element_id_template: str = \"progress-span-{job_id}\", # Template for generating element ID\n initial_content=None)\n \"Create an SSE-enabled progress element.\n\nArgs:\n job_id: Job identifier\n endpoint_template: Template for SSE endpoint URL\n element_id_template: Template for element ID\n initial_content: Initial content to display\n \nReturns:\n SSE-configured element for progress updates\"\n \n def sse_status_element(job_id: str,\n endpoint_template: str = \"/stream_job_status?job_id={job_id}\", # URL template for the SSE endpoint\n element_id_template: str = \"status-span-{job_id}\", # Template for generating element ID\n initial_content=None)\n \"Create an SSE-enabled status element.\n\nArgs:\n job_id: Job identifier\n endpoint_template: Template for SSE endpoint URL\n element_id_template: Template for element ID\n initial_content: Initial content to display\n \nReturns:\n SSE-configured element for status updates\"\n \n def create_sse_monitor_script(\n config: Dict[str, Any] # Configuration dictionary for monitoring setup\n )\n \"Create a monitoring script for SSE connections.\n\nArgs:\n config: Configuration dictionary with keys:\n - sse_element_id: ID of SSE element to monitor\n - status_element_id: ID of status display element\n - auto_reconnect: Whether to auto-reconnect\n - debug: Whether to enable debug logging\n - status_indicators: Dict of status HTML strings\n \nReturns:\n Script element with monitoring code\"\n```\n\n### Connection monitoring & config (`monitoring.ipynb`)\n\n> Connection monitoring and debugging tools for SSE applications.\n> Provides configurable status indicators, automatic reconnection, and\n> visibility change handling.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_sse.monitoring import (\n SSEMonitorConfig,\n create_sse_monitor\n)\n```\n\n#### Functions\n\n``` python\ndef create_sse_monitor(\n htmx_sse: HTMXSSEConnector,\n config: SSEMonitorConfig # SSEMonitorConfig instance\n)\n \"Create a connection monitor with the specified configuration.\"\n```\n\n#### Classes\n\n``` python\n@dataclass\nclass SSEMonitorConfig:\n \"Configuration for SSE connection monitoring.\"\n \n sse_element_id: str = 'sse-connection'\n status_element_id: str = 'connection-status'\n auto_reconnect: bool = True\n reconnect_delay: int = 3000\n debug: bool = False\n heartbeat_timeout: int = 30000\n status_indicators: Optional[Dict[str, str]]\n```\n\n### SSEElementUpdater (`updater.ipynb`)\n\n> Flexible element update system for building out-of-band (OOB) swap\n> elements. Register handlers by event type and compose updates without\n> coupling to specific UI components.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_sse.updater import (\n SSEElementUpdater\n)\n```\n\n#### Classes\n\n``` python\nclass SSEElementUpdater:\n def __init__(self):\n \"\"\"Initialize the SSE Element Updater.\"\"\"\n self._handlers: Dict[str, List[Callable]] = {}\n \"\"\"\n Builds OOB swap elements without hardcoding UI components.\n This class provides a flexible system for registering and executing\n element update handlers based on event types.\n \"\"\"\n \n def __init__(self):\n \"\"\"Initialize the SSE Element Updater.\"\"\"\n self._handlers: Dict[str, List[Callable]] = {}\n \"Initialize the SSE Element Updater.\"\n \n def register(\n self,\n event_type: str, # The event type to handle\n priority: int = 0 # Handler priority (higher numbers run first)\n ): # Decorator function\n \"Decorator to register an update handler for a specific event type.\"\n \n def register_handler(\n self,\n event_type: str, # The event type to handle\n handler: Callable, # The handler function\n priority: int = 0 # Handler priority (higher numbers run first)\n )\n \"Register an update handler programmatically.\"\n \n def set_default_handler(\n self,\n handler: Callable # The default handler function\n )\n \"Set a default handler for unregistered event types.\"\n \n def add_preprocessor(\n self,\n processor: Callable # Function that processes (event_type, data) and returns modified data\n )\n \"Add a preprocessor that runs before handlers.\"\n \n def add_postprocessor(\n self,\n processor: Callable # Function that processes elements list and returns modified elements\n )\n \"Add a postprocessor that runs after handlers.\"\n \n def create_elements(\n self,\n event_type: str, # The type of event\n data: Dict[str, Any] # Event data\n ) -> List[Any]: # List of elements to be sent via SSE\n \"Create elements for a given event type and data.\"\n \n def clear_handlers(\n self,\n event_type: Optional[str] = None # Optional specific event type to clear\n )\n \"Clear handlers for a specific event type or all handlers.\"\n \n def get_registered_events(\n self\n ) -> List[str]: # List of event types with registered handlers\n \"Get list of registered event types.\"\n```\n",
"bugtrack_url": null,
"license": "Apache Software License 2.0",
"summary": "Real-time Server-Sent Events (SSE) and HTMX integration library for FastHTML with cross-tab synchronization support.",
"version": "0.0.20",
"project_urls": {
"Homepage": "https://github.com/cj-mills/cjm-fasthtml-sse"
},
"split_keywords": [
"nbdev",
"jupyter",
"notebook",
"python"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "e2d58622e8b3e0849c4eec2014725738d96df966dbcfcd1953831dd36020c6ca",
"md5": "b2f1f83ea3b7c44654d40fb850e26c12",
"sha256": "ca958c7b5c0467c56672208346bf2bf322fe3ac53d17e3ac79ef038d68a99973"
},
"downloads": -1,
"filename": "cjm_fasthtml_sse-0.0.20-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b2f1f83ea3b7c44654d40fb850e26c12",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 22605,
"upload_time": "2025-08-29T19:33:41",
"upload_time_iso_8601": "2025-08-29T19:33:41.612901Z",
"url": "https://files.pythonhosted.org/packages/e2/d5/8622e8b3e0849c4eec2014725738d96df966dbcfcd1953831dd36020c6ca/cjm_fasthtml_sse-0.0.20-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "b011e263d74d1a5ab389f4c63bf14b3ddb8283d2aac6e84911802ab9b17e6162",
"md5": "50d847aa1db435494ca595dd1d31b553",
"sha256": "400941197641c5addc3972bcd7bb5edcd43a1ca6e1dce4257e1422ee71f0a821"
},
"downloads": -1,
"filename": "cjm_fasthtml_sse-0.0.20.tar.gz",
"has_sig": false,
"md5_digest": "50d847aa1db435494ca595dd1d31b553",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 24559,
"upload_time": "2025-08-29T19:33:43",
"upload_time_iso_8601": "2025-08-29T19:33:43.744423Z",
"url": "https://files.pythonhosted.org/packages/b0/11/e263d74d1a5ab389f4c63bf14b3ddb8283d2aac6e84911802ab9b17e6162/cjm_fasthtml_sse-0.0.20.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-29 19:33:43",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cj-mills",
"github_project": "cjm-fasthtml-sse",
"github_not_found": true,
"lcname": "cjm-fasthtml-sse"
}
JSON
