yera.dev.services.session_service

Session service for managing app sessions.

Symbols

class SessionService — Service for managing app sessions and generating SSE events.

SessionService

Service for managing app sessions and generating SSE events.

Methods

generate_session_id — Generate a unique session ID.
create_session — Create a new session for the app.
get_session — Retrieve an existing session.
add_interaction — Add an interaction to a session.
generate_events — Generate SSE events from Python app execution.
cleanup_session — Clean up session-specific resources.
cleanup_all — Clean up all sessions.

SessionService.generate_session_id

generate_session_id() → str

Generate a unique session ID.

Returns

type: str

UUID string for the session

SessionService.create_session

create_session(
    app: AppFunctionWrapper,
    args: tuple = (),
    kwargs: dict | None = None,
) → str

Create a new session for the app.

Parameters

app
type: AppFunctionWrapper

The app wrapper to execute.

args
type: tuple = ()

Positional arguments to pass to the app's invoke method.

kwargs
type: dict | None = None

Keyword arguments to pass to the app's invoke method.

Returns

type: str

Generated session_id

SessionService.get_session

get_session(
    session_id: str,
) → SessionState

Retrieve an existing session.

Parameters

session_id
type: str

Session identifier

Returns

type: SessionState

Python app session state

Raises

KeyError

If session not found

SessionService.add_interaction

add_interaction(
    session_id: str,
    interaction: dict,
) → None

Add an interaction to a session.

Converts the UserInteraction to an InputEvent and pushes it to the executor's EventStream input queue. Also stores the interaction in session state for debugging.

Parameters

session_id
type: str

Session identifier

interaction
type: dict

Interaction dictionary (UserInteraction format)

Raises

KeyError

If session not found

RuntimeError

If executor not started

SessionService.generate_events

generate_events(
    session_id: str,
) → AsyncIterator[str]

Generate SSE events from Python app execution.

SessionService.cleanup_session

cleanup_session(
    session_id: str,
) → None

Clean up session-specific resources.

Stops the executor process if running and removes the session from internal storage.

Parameters

session_id
type: str

Session identifier to clean up

Raises

KeyError

If session not found

SessionService.cleanup_all

cleanup_all() → None

Clean up all sessions.

Iterates through all sessions and calls cleanup_session() for each. Called on server shutdown to ensure no orphaned processes remain.