cjm-fasthtml-workflow-session-management

Install

pip install cjm_fasthtml_workflow_session_management

Project Structure

nbs/
├── components/ (3)
│   ├── helpers.ipynb        # Shared rendering helpers for the workflow session management interface.
│   ├── page_renderer.ipynb  # Full session management page assembly — header with title and the session list body.
│   └── session_list.ipynb   # Session list component with virtual collection integration, keyboard navigation, and action modals.
├── routes/ (2)
│   ├── init.ipynb      # Top-level router assembly for the session management library.
│   └── sessions.ipynb  # Session management route handlers — list, create, delete, rename, and resume.
├── services/ (1)
│   └── management.ipynb  # Service layer wrapping `cjm-workflow-state` for session management operations
├── html_ids.ipynb  # HTML ID constants for the workflow session management interface
├── models.ipynb    # Data models for the workflow session management interface
└── utils.ipynb     # Formatting utilities for the session management interface

Total: 9 notebooks across 3 directories

Module Dependencies

graph LR
    components_helpers[components.helpers<br/>components.helpers]
    components_page_renderer[components.page_renderer<br/>components.page_renderer]
    components_session_list[components.session_list<br/>components.session_list]
    html_ids[html_ids<br/>html_ids]
    models[models<br/>Models]
    routes_init[routes.init<br/>routes.init]
    routes_sessions[routes.sessions<br/>routes.sessions]
    services_management[services.management<br/>services.management]
    utils[utils<br/>utils]

    components_page_renderer --> html_ids
    components_page_renderer --> components_helpers
    components_page_renderer --> models
    components_session_list --> html_ids
    components_session_list --> utils
    components_session_list --> models
    components_session_list --> components_helpers
    routes_init --> routes_sessions
    routes_init --> html_ids
    routes_init --> components_session_list
    routes_init --> components_page_renderer
    routes_init --> models
    routes_init --> services_management
    routes_sessions --> models
    routes_sessions --> services_management
    services_management --> utils
    services_management --> models

17 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

components.helpers (helpers.ipynb)

Shared rendering helpers for the workflow session management interface.

Import

from cjm_fasthtml_workflow_session_management.components.helpers import (
    DEBUG_SESSION_RENDER,
    render_section_header,
    render_icon_button,
    render_alert,
    render_active_session_badge
)

Functions

def render_section_header(
    title:str, # Section title text
    icon_name:str, # Lucide icon name (kebab-case)
) -> Any: # Header element with icon and title
    "Render a section header with a leading lucide icon."

def render_icon_button(
    icon_name:str, # Lucide icon name (kebab-case)
    label:str, # Accessible tooltip label
    color:Optional[str]=None, # DaisyUI button color class (e.g. btn_colors.error)
    size:Optional[str]=None, # DaisyUI button size class (e.g. btn_sizes.sm)
    **kwargs # Additional HTML attributes (hx_post, onclick, etc.)
) -> Any: # Button element with icon
    """
    Render a ghost-style button with an icon and accessible label.
    
    Defaults to `type="button"` to prevent accidental form submission when
    rendered inside a wrapping form. Callers can override by passing `type=...`.
    """

def render_alert(
    message:str, # Alert message text
    color:Optional[str]=None, # DaisyUI alert color class (e.g. alert_colors.success)
    alert_id:str="", # Optional HTML ID for the alert
) -> Any: # Alert element
    "Render a DaisyUI alert message."

def render_active_session_badge(
    label:str="Active", # Badge label text
    badge_id:str="", # Optional HTML ID for the badge
) -> Any: # Span element containing the badge
    "Render the "Active" indicator badge for the currently active session row."

Variables

DEBUG_SESSION_RENDER = False  # Enable for verbose component rendering logs

html_ids (html_ids.ipynb)

HTML ID constants for the workflow session management interface

Import

from cjm_fasthtml_workflow_session_management.html_ids import (
    SessionManagerHtmlIds
)

Classes

class SessionManagerHtmlIds:
    "HTML ID constants for the workflow session management interface."
    
    def as_selector(
            id_str:str # The HTML ID to convert
        ) -> str: # CSS selector with # prefix
        "Convert an ID to a CSS selector format."

routes.init (init.ipynb)

Top-level router assembly for the session management library.

Import

from cjm_fasthtml_workflow_session_management.routes.init import (
    init_session_manager_routers
)

Functions

def init_session_manager_routers(
    service:SessionManagementService, # Service for session CRUD
    workflow_url:str, # URL to redirect to on resume (e.g., the workflow root)
    prefix:str="/manage/sessions", # Base prefix for all session management routes
    column_specs:Optional[List[ColumnSpec]]=None, # Host enricher column specs
    get_step_title:Optional[Callable[[str], str]]=None, # Optional step ID -> human title mapper
    page_title:str="Sessions", # Page header title
    page_icon:str="layers", # Lucide icon for the page header
) -> SessionManagementResult: # Assembled result with routers, urls, and render callables
    "Initialize all session management routers with virtual collection integration."

services.management (management.ipynb)

Service layer wrapping cjm-workflow-state for session management operations

Import

from cjm_fasthtml_workflow_session_management.services.management import (
    DEBUG_SESSION_SERVICE,
    SessionManagementService
)

Functions

@patch
def list_sessions(
    self:SessionManagementService,
    order_by:str="updated_at", # Sort column: "updated_at", "created_at", or "label"
    descending:bool=True # Sort direction
) -> List[EnrichedSessionSummary]: # All sessions with labels resolved and enricher applied
    "List all sessions for the managed flow with display enrichment applied."

@patch
def get_session(
    self:SessionManagementService,
    session_id:str # Session identifier
) -> Optional[EnrichedSessionSummary]: # Enriched session, or None if not found
    "Fetch a single enriched session summary by ID."

@patch
def session_exists(
    self:SessionManagementService,
    session_id:str # Session identifier
) -> bool: # True if a row exists
    "Check whether a session exists in the managed flow."

@patch
def create_session(
    self:SessionManagementService,
    label:Optional[str]=None # Optional human-readable label for the new session
) -> str: # The new session_id (auto-generated UUID4)
    "Create a new empty session and return its session_id."

@patch
def rename_session(
    self:SessionManagementService,
    session_id:str, # Session identifier
    label:Optional[str] # New label, or None to clear
) -> None
    "Update a session's label. No-op if the session does not exist."

@patch
def delete_session(
    self:SessionManagementService,
    session_id:str # Session identifier
) -> None
    "Delete a session row. Idempotent."

Classes

class SessionManagementService:
    def __init__(
        self,
        state_store:SQLiteWorkflowStateStore, # The workflow state store to manage sessions in
        flow_id:str, # Workflow identifier this service is scoped to
        enricher:Optional[SessionEnricher]=None, # Optional host callback that enriches list display
        label_generator:Optional[LabelGenerator]=None, # Optional host callback for default labels
    )
    "Service wrapping cjm-workflow-state for session listing and lifecycle operations."
    
    def __init__(
            self,
            state_store:SQLiteWorkflowStateStore, # The workflow state store to manage sessions in
            flow_id:str, # Workflow identifier this service is scoped to
            enricher:Optional[SessionEnricher]=None, # Optional host callback that enriches list display
            label_generator:Optional[LabelGenerator]=None, # Optional host callback for default labels
        )
        "Initialize with a state store, flow ID, and optional host callbacks."
    
    def flow_id(self) -> str: # The flow ID this service manages
            """The workflow identifier this service is scoped to."""
            return self._flow_id
        
        def _resolve_label(
            self,
            summary:SessionSummary, # Raw session metadata
            state_json:Dict[str, Any] # Full state blob for label generator input
        ) -> str: # Resolved display label
        "The workflow identifier this service is scoped to."

Variables

DEBUG_SESSION_SERVICE = False  # Enable for verbose session service logging

Models (models.ipynb)

Data models for the workflow session management interface

Import

from cjm_fasthtml_workflow_session_management.models import (
    SessionEnricher,
    LabelGenerator,
    EnrichedSessionSummary,
    ColumnSpec,
    SessionManagementUrls,
    SessionManagementResult
)

Classes

@dataclass
class EnrichedSessionSummary:
    "Session metadata plus resolved label and host-enriched display fields."
    
    summary: SessionSummary  # Raw session metadata from cjm-workflow-state
    resolved_label: str  # summary.label if set, else label_generator output, else a default
    enriched_fields: Dict[str, str] = field(...)  # Host enricher output — keyed by column field name

@dataclass
class ColumnSpec:
    "Declarative spec for a single enricher-backed column in the session list."
    
    field: str  # Key into EnrichedSessionSummary.enriched_fields
    header: str  # Human-readable column header
    width_class: Optional[str]  # Optional tailwind width class (e.g., "w-24") for the column

@dataclass
class SessionManagementUrls:
    "URL bundle for session management route endpoints."
    
    management_page: str  # GET: full session manager page
    list_sessions: str  # GET: session list fragment (for OOB refresh)
    session_detail: str  # GET: + ?session_id=... for detail view
    create_session: str  # POST: mint a new session and switch to it
    delete_session: str  # POST: + session_id in form data
    rename_session: str  # POST: + session_id and new label
    resume_session: str  # POST: set_session_id + redirect to workflow_url

@dataclass
class SessionManagementResult:
    "Result of session manager router initialization."
    
    routers: List[Any]  # APIRouter instances to register
    urls: SessionManagementUrls  # URL bundle for route endpoints
    routes: Dict[str, Callable]  # Route handler functions keyed by name
    render_page: Callable  # () -> full session manager page component
    render_list: Callable  # () -> session list component
    refresh_items: Callable  # async () -> refresh items from the service
    resume_session: Callable[[Any, str], None]  # (sess, session_id) -> set active session ID in the HTTP session

components.page_renderer (page_renderer.ipynb)

Full session management page assembly — header with title and the session list body.

Import

from cjm_fasthtml_workflow_session_management.components.page_renderer import (
    render_page_header,
    render_session_manager_page
)

Functions

def render_page_header(
    title:str="Sessions", # Page title
    icon_name:str="layers", # Lucide icon for the title
) -> Any: # Header element
    "Render the page header with an icon + title."

def render_session_manager_page(
    urls:SessionManagementUrls, # URL bundle (unused directly, reserved for future header actions)
    render_list_fn:Callable, # () -> session list component
    title:str="Sessions", # Page title
    icon_name:str="layers", # Lucide icon name for the title
) -> Any: # Complete session manager page component
    "Render the complete session manager page with header and session list."

components.session_list (session_list.ipynb)

Session list component with virtual collection integration, keyboard navigation, and action modals.

Import

from cjm_fasthtml_workflow_session_management.components.session_list import (
    build_session_columns,
    create_session_cell_renderer,
    render_session_toolbar,
    render_rename_modal,
    render_list_scripts,
    render_session_list
)

Functions

def build_session_columns(
    column_specs:Optional[List[ColumnSpec]]=None, # Host-supplied enricher column specs
) -> tuple: # Tuple of ColumnDef for the VC config
    """
    Build the column definitions for the session list.
    
    Produces a fixed column layout with host enricher columns inserted between
    status and updated columns. If `column_specs` is None or empty, only fixed
    columns are included.
    """

def create_session_cell_renderer(
    urls:SessionManagementUrls, # URL bundle for action buttons
    get_active_session_id:Callable[[], str], # () -> currently active session ID
    get_step_title:Optional[Callable[[str], str]]=None, # Optional step ID -> human title mapper
) -> Callable: # render_cell(item, ctx) -> FT component
    "Create a cell renderer for the session list virtual collection."

def render_session_toolbar(
    urls:SessionManagementUrls, # URL bundle for action routes
    total_count:int=0, # Total number of sessions in the list
) -> Any: # Toolbar element
    "Render the session list toolbar with session count and New Session button."

def render_rename_modal(
    urls:SessionManagementUrls, # URL bundle for rename route
    modal_id:str=SessionManagerHtmlIds.RENAME_MODAL, # Dialog element ID
    input_id:str=SessionManagerHtmlIds.RENAME_INPUT, # Input element ID
) -> Any: # Dialog element
    "Render the rename session modal using HTML5 dialog with backdrop click-to-close."

def render_list_scripts(
    urls:SessionManagementUrls, # URL bundle for action routes
) -> Any: # Script element
    "Render client-side JavaScript for delete and rename modal management."

def render_session_list(
    items:list, # Current session list (List[EnrichedSessionSummary])
    vc_config:VirtualCollectionConfig, # VC configuration
    vc_state:VirtualCollectionState, # VC state
    vc_ids:VirtualCollectionHtmlIds, # VC HTML IDs
    vc_btn_ids:VirtualCollectionButtonIds, # VC button IDs
    vc_urls:VirtualCollectionUrls, # VC route URLs
    mgmt_urls:SessionManagementUrls, # Session manager URLs
    render_cell:Callable, # Cell renderer callback
) -> Any: # Complete session list component
    "Render the session list with virtual collection, keyboard nav, and modals."

routes.sessions (sessions.ipynb)

Session management route handlers — list, create, delete, rename, and resume.

Import

from cjm_fasthtml_workflow_session_management.routes.sessions import (
    DEBUG_SESSION_ROUTES,
    init_session_router
)

Functions

def init_session_router(
    service:SessionManagementService, # Service for session CRUD
    prefix:str, # Route prefix (e.g., "/manage/sessions")
    urls:SessionManagementUrls, # URL bundle (populated by caller after init)
    workflow_url:str, # Where to redirect after resume
    refresh_items:Callable, # (request=None) -> reload items from service
    refresh_items_oob:Callable, # (request=None) -> refresh + OOB tuple for HTMX response
    render_page:Callable, # () -> full session manager page
    render_list:Callable, # () -> session list component
) -> Tuple[APIRouter, Dict[str, Callable]]: # (router, routes dict)
    "Initialize session management routes (list, create, delete, rename, resume)."

Variables

DEBUG_SESSION_ROUTES = False  # Enable for verbose session route logging

utils (utils.ipynb)

Formatting utilities for the session management interface

Import

from cjm_fasthtml_workflow_session_management.utils import (
    parse_sqlite_timestamp,
    format_relative_time,
    format_absolute_datetime,
    format_bytes,
    default_label
)

Functions

def parse_sqlite_timestamp(
    ts:Optional[str] # ISO timestamp string from SQLite, e.g. "2026-04-08 19:11:50"
) -> Optional[datetime]: # Parsed UTC datetime, or None on failure
    "Parse a SQLite CURRENT_TIMESTAMP string into a timezone-aware UTC datetime."

def format_relative_time(
    ts:Optional[str], # SQLite ISO timestamp string
    now:Optional[datetime]=None # Reference "now" for testing; defaults to real now in UTC
) -> str: # Short human-readable relative time
    "Format a SQLite timestamp as a relative time string like "2 hours ago"."

def format_absolute_datetime(
    ts:Optional[str] # SQLite ISO timestamp string
) -> str: # Formatted datetime, e.g. "Apr 08, 2026 19:11"
    "Format a SQLite timestamp as an absolute human-readable datetime."

def format_bytes(
    n:Optional[int] # Size in bytes
) -> str: # Compact human-readable size, e.g. "1.2 KB"
    "Format a byte count as a compact human-readable size."

def default_label(
    created_at:Optional[str] # SQLite ISO timestamp of session creation
) -> str: # Fallback label, e.g. "Session 2026-04-08 19:11"
    "Produce a fallback session label based on creation time."