# 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]
helpers --> htmx
monitoring --> htmx
```
*2 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."
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 broadcast manager with connection pooling and message history."
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 event dispatcher with empty handler registry."""
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 event dispatcher with empty handler registry."""
self._handlers: Dict[str, List[tuple[int, Callable]]] = {}
"Initialize the event dispatcher with empty handler registry."
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(
htmx_sse: HTMXSSEConnector,
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."
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."
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."
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."
def create_sse_monitor_script(
config: Dict[str, Any] # Configuration dictionary for monitoring setup
) -> FT: # Script element with monitoring code
"Create a monitoring script for SSE connections."
```
### 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
) -> FT: # Script element with monitoring code
"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 updater with empty handler registry."""
self._handlers: Dict[str, List[Callable]] = {}
"Builds OOB swap elements without hardcoding UI components."
def __init__(self):
"""Initialize the updater with empty handler registry."""
self._handlers: Dict[str, List[Callable]] = {}
"Initialize the updater with empty handler registry."
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/58/17/c621fa6b3661c63d978edad5b3f76fe7f52f57306878ced9814867a03696/cjm_fasthtml_sse-0.0.23.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 helpers --> htmx\n monitoring --> htmx\n```\n\n*2 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 \"Manages SSE connections and broadcasting without UI dependencies.\"\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 broadcast manager with connection pooling and message history.\"\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 event dispatcher with empty handler registry.\"\"\"\n self._handlers: Dict[str, List[tuple[int, Callable]]] = {}\n \"Decoupled event routing system with namespace support, middleware, filtering, and priority-based handling.\"\n \n def __init__(self):\n \"\"\"Initialize the event dispatcher with empty handler registry.\"\"\"\n self._handlers: Dict[str, List[tuple[int, Callable]]] = {}\n \"Initialize the event dispatcher with empty handler registry.\"\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(\n htmx_sse: HTMXSSEConnector,\n 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 \"Provides helper functions for setting up HTMX SSE connections without hardcoding specific implementations.\"\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 \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 \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 \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 \n def create_sse_monitor_script(\n config: Dict[str, Any] # Configuration dictionary for monitoring setup\n ) -> FT: # Script element with monitoring code\n \"Create a monitoring script for SSE connections.\"\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) -> FT: # Script element with monitoring code\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 updater with empty handler registry.\"\"\"\n self._handlers: Dict[str, List[Callable]] = {}\n \"Builds OOB swap elements without hardcoding UI components.\"\n \n def __init__(self):\n \"\"\"Initialize the updater with empty handler registry.\"\"\"\n self._handlers: Dict[str, List[Callable]] = {}\n \"Initialize the updater with empty handler registry.\"\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.23",
"project_urls": {
"Homepage": "https://github.com/cj-mills/cjm-fasthtml-sse"
},
"split_keywords": [
"nbdev",
"jupyter",
"notebook",
"python"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "0fd117d5fb8ea713804887f56b3eecd59ba1797e46d90abe41ad7418ee497839",
"md5": "109a6f39c73b0890be47806bb0e4cf76",
"sha256": "e1a7bfffa99cc294d855eca89c2a178f031e317a609b63067dd1f21f8e543b9c"
},
"downloads": -1,
"filename": "cjm_fasthtml_sse-0.0.23-py3-none-any.whl",
"has_sig": false,
"md5_digest": "109a6f39c73b0890be47806bb0e4cf76",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 21678,
"upload_time": "2025-10-24T22:29:48",
"upload_time_iso_8601": "2025-10-24T22:29:48.495854Z",
"url": "https://files.pythonhosted.org/packages/0f/d1/17d5fb8ea713804887f56b3eecd59ba1797e46d90abe41ad7418ee497839/cjm_fasthtml_sse-0.0.23-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "5817c621fa6b3661c63d978edad5b3f76fe7f52f57306878ced9814867a03696",
"md5": "a4f7906c4b9bbc3ff54b1bf2fca8615c",
"sha256": "2e1f2a00a113392f649572147c74ac20bcef4c07a2aa3a0c2085b58eaa5a9f5e"
},
"downloads": -1,
"filename": "cjm_fasthtml_sse-0.0.23.tar.gz",
"has_sig": false,
"md5_digest": "a4f7906c4b9bbc3ff54b1bf2fca8615c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 23154,
"upload_time": "2025-10-24T22:29:49",
"upload_time_iso_8601": "2025-10-24T22:29:49.700905Z",
"url": "https://files.pythonhosted.org/packages/58/17/c621fa6b3661c63d978edad5b3f76fe7f52f57306878ced9814867a03696/cjm_fasthtml_sse-0.0.23.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-24 22:29:49",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cj-mills",
"github_project": "cjm-fasthtml-sse",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "cjm-fasthtml-sse"
}
JSON
