Public library (sparkth.lib)

sparkth/lib/ is the curated, stable API that application code and plugins import from, instead of reaching into sparkth.core.* (or sparkth.llm.* / sparkth.rag.*) directly (see #379). Everything below is generated from the module docstrings.

The plugin authoring surface and the permissions API have their own pages: Plugin authoring and Permissions.

Database sessions

sparkth.lib.db

Database session access for Sparkth — the curated public session API.

This is the single public entry point for obtaining a database session. All code, including plugins, should acquire sessions from here rather than reaching for the raw SQLAlchemy engine in :mod:sparkth.core.db or :mod:sparkth.core.analytics.db.

The engine itself stays in :mod:sparkth.core.db (app DB) and :mod:sparkth.core.analytics.db (analytics DB); this module is only the public face that hands out sessions over them.

analytics_session_scope async

analytics_session_scope(
    expire_on_commit: bool = False,
) -> AsyncGenerator[AsyncSession, None]

Open an async session against the analytics database as a managed context.

Identical in contract to :func:session_scope, but bound to the analytics engine (the separate analytics database). Use this for background/non-request analytics code (event ingestion, rollup maintenance). Inside request handlers, prefer the :func:get_analytics_session dependency.

Delegates to analytics_db.open_analytics_session — the same seam that the test suite overrides to inject a throwaway engine.

Parameters:

Name	Type	Description	Default
expire_on_commit	bool	Whether ORM objects are expired after commit(). Defaults to False (async-safe).	False

Yields:

Name	Type	Description
An	AsyncGenerator[AsyncSession, None]	class:AsyncSession bound to the analytics async engine.

get_analytics_session async

get_analytics_session() -> AsyncGenerator[
    AsyncSession, None
]

FastAPI dependency providing an :class:AsyncSession bound to the analytics database.

Delegates to :func:analytics_session_scope. Parameterless for the same reason as :func:get_async_session.

get_async_session async

get_async_session() -> AsyncGenerator[AsyncSession, None]

FastAPI dependency that provides an :class:AsyncSession to async def handlers.

Delegates to :func:session_scope. It is intentionally parameterless: FastAPI turns a dependency's parameters into request (query) parameters, so the expire_on_commit knob must not be exposed here — use :func:session_scope directly when you need to override it.

session_scope async

session_scope(
    expire_on_commit: bool = False,
) -> AsyncGenerator[AsyncSession, None]

Open an async database session as a managed context.

Yields an :class:AsyncSession — a unit-of-work that borrows a connection from the shared engine (sparkth.core.db.get_engine), tracks the ORM objects you load and mutate, and returns the connection to the pool when the async with block exits (whether normally or via an exception). Always use it as a context manager so the connection is never leaked::

from sparkth.lib.db import session_scope

async with session_scope() as session:
    session.add(obj)
    await session.commit()

Use this for code that runs outside an HTTP request — background tasks, plugin bootstrap, CLI jobs, cleanup routines — where FastAPI's dependency injection is not available. Inside request handlers, prefer the :func:get_async_session dependency instead.

Transaction semantics: the caller is responsible for committing (await session.commit()); any un-committed work is rolled back when the context exits.

The expire_on_commit parameter defaults to False, which is the async-safe choice. By default SQLAlchemy expires every ORM object after commit(), so the next attribute access lazily re-issues a SELECT to reload it. In synchronous code that reload is a transparent (if wasteful) blocking query, but in async code it is implicit I/O that cannot be awaited and fails once the session has closed. Keeping expire_on_commit=False leaves already-loaded attributes valid after the commit and after the block. Pass expire_on_commit=True only if you specifically want post-commit objects to refresh on next access.

Parameters:

Name	Type	Description	Default
expire_on_commit	bool	Whether ORM objects are expired after commit(). Defaults to False (async-safe).	False

Yields:

Name	Type	Description
An	AsyncGenerator[AsyncSession, None]	class:AsyncSession bound to the engine from :func:sparkth.core.db.get_engine.

Logging

sparkth.lib.log

Centralized logging for Sparkth.

The single public entry point for logging. All modules — application code and plugins alike — must obtain loggers via :func:get_logger, never logging.getLogger directly. Logging is configured exactly once per process via :func:configure_logging, which is the only logging.basicConfig call in the codebase.

Example

from sparkth.lib.log import get_logger

logger = get_logger(__name__)
logger.info("This is a log message")

configure_logging

configure_logging(level: int = logging.INFO) -> None

Configure root logging once for the whole process.

This is the only place in the codebase that calls logging.basicConfig. It is idempotent: basicConfig is a no-op if the root logger already has handlers, so calling it from multiple entrypoints is safe.

Parameters:

Name	Type	Description	Default
level	int	Root logging level (defaults to INFO).	INFO

get_logger

get_logger(name: str) -> logging.Logger

Return the Sparkth logger for name (typically __name__).

Loggers are namespaced under sparkth and propagate to the root logger configured by :func:configure_logging. Application modules pass __name__, which already starts with sparkth (the package name), so it is used as-is; any other name is prefixed with sparkth. to keep every logger under the single sparkth root.

Parameters:

Name	Type	Description	Default
name	str	Logger name, typically __name__ of the calling module.	required

Returns:

Type	Description
Logger	Logger instance.

Settings

sparkth.lib.settings

Application settings public API for Sparkth.

The single public entry point for application settings. All modules — application code and plugins alike — must access settings via :func:get_settings, never by importing from sparkth.core.config directly.

A plugin defining its own BaseSettings class must pass :data:ENV_FILES as its env_file so it reads the same env files, in the same precedence order, as the core Settings class.

Example

from sparkth.lib.settings import get_settings

settings = get_settings()
print(settings.SECRET_KEY)

Language

sparkth.lib.language

Public API for language resolution and language naming.

The single entry point for the supported-language allowlist, for checking whether a language tag belongs to it, and for naming an arbitrary language tag for a prompt. Application code and plugins import from here, never from sparkth.core.language or sparkth.core.config.

Example

from sparkth.lib.language import is_supported_language, language_display_name

if current_user.language and is_supported_language(current_user.language):
    ...   # membership check, e.g. before binding
name = language_display_name("pt-BR")   # "Portuguese (Brazil)", for a prompt

LanguageInfo

Bases: NamedTuple

Display names for a supported language.

Attributes:

Name	Type	Description
name	str	The language's name in English, for logs and admin surfaces.
native_name	str	The endonym, for the user-facing picker — someone looking for their language reads it in their own language.

is_supported_language

is_supported_language(tag: str) -> bool

Whether tag is one of the platform's supported BCP 47 tags.

The single expression of allowlist membership: DEFAULT_LANGUAGE validation and every other caller share it, so the rule cannot drift between what the platform accepts as its default and what it accepts from a user. Matching is exact and case-sensitive — en-US and EN are unsupported rather than normalised to en.

language_display_name

language_display_name(tag: str | None) -> str

The English name of the language tag identifies.

tag is an arbitrary BCP 47 tag supplied by a caller — it is deliberately not checked against :data:SUPPORTED_LANGUAGES, which governs the interface translations the platform ships rather than the languages the model may write in. "de" therefore yields "German" even though no German interface exists.

Falls back to the platform default's name when tag is absent, is not a parseable language tag, or parses to a locale with no English display name in CLDR (e.g. "skr", Saraiki) — so a misspelled or obscure value degrades to the default instead of failing the caller's whole request or surfacing the literal string "None". Babel's parser defaults to the POSIX underscore separator, so the BCP 47 hyphen is passed explicitly.

resolve_language

resolve_language(tag: str | None) -> str

The language a stored preference of tag resolves to.

tag is a user's stored choice, or None when they never made one. Returns it when it is still supported, and otherwise DEFAULT_LANGUAGE — which covers both cases where the stored value cannot be honoured: the user never chose, and the tag has since left the allowlist. A language withdrawn from the list (say, because its shipped interface translation was pulled) must stop being handed back for the users who had already picked it.

Internationalization

sparkth.lib.i18n

Static-translation (i18n) public API for Sparkth.

The single public entry point for translating user-facing strings. All modules — application code and plugins alike — must import the marking functions from here, never from sparkth.core.i18n directly.

How to mark a string:

• Inside request handling, wrap the literal in :func:gettext, imported as _::

  from sparkth.lib.i18n import _

  raise HTTPException(status_code=401, detail=_("Incorrect username or password"))

• f-strings cannot be extracted by pybabel; convert them to str.format on the translated template::

  _("Role not found: {role_name}").format(role_name=role_name)

• Module-level constants evaluate at import, before any request locale exists; mark them with :func:lazy_gettext and render with str() at the boundary::

  from sparkth.lib.i18n import lazy_gettext

  GREETING_MESSAGE = lazy_gettext("Hello! How can I help you?")

• A constant stored in a plain-str field (a dataclass a response model embeds, where a :class:LazyString cannot go): mark the literal with :func:gettext_noop at the definition and translate the stored value by passing it through :func:gettext where it is rendered::

  DisplayInfo(gettext_noop("Create Course"), gettext_noop("Build courses with AI"))

Extraction and catalogs are driven by the i18n.* Make targets (see make help). Core's catalogs live in sparkth/locale; a plugin ships its own by registering its catalog directory on the :data:LOCALE_DIRS hook from its __init__::

from pathlib import Path

from sparkth.lib.i18n import LOCALE_DIRS

LOCALE_DIRS.add_item(Path(__file__).parent / "locale")

The locale itself is negotiated per request from Accept-Language by the locale middleware and read via :func:get_locale. The allowlist that negotiation matches against, and the resolution of a user's stored preference, belong to :mod:sparkth.lib.language; this module covers translation only.

LazyString

LazyString(message: str)

A translation deferred to render time.

Module-level constants evaluate at import, before any request locale exists; wrapping them keeps the marking at the definition site while the translation happens when the value is rendered. The object is not a str — call str() on it at the boundary (response serialization, message dispatch).

bind_locale

bind_locale(tag: str) -> None

Install tag as the active locale for the rest of the request.

Mirrors :func:sparkth.core.audit.context.bind_audit_actor: the locale middleware runs before authentication, so it can only negotiate Accept-Language. Once a signed-in user is resolved, their stored choice is the better answer and replaces the negotiated one — they chose it, and their browser did not.

Deliberately not a context manager: there is nothing to restore, because the request's context is discarded when its task ends. This relies on the caller running in the same task as the route, which holds for an async FastAPI dependency but not for a sync one — a sync dependency runs in a threadpool with a copied context and the binding would be lost.

get_locale

get_locale() -> str

Return the active locale, or DEFAULT_LANGUAGE when none is bound.

gettext

gettext(message: str) -> str

Translate message into the active request locale.

The canonical marker for user-facing strings; import it as _ so pybabel extract picks the call sites up.

An empty message is returned as-is: a compiled catalog stores its metadata header under the empty msgid, so looking one up would return the PO header instead. Callers reach this with an empty string through stored source messages (a gettext_noop-marked field left blank), not from a literal.

gettext_noop

gettext_noop(message: str) -> str

Mark message for extraction and return it unchanged.

For English strings stored in plain-str fields (e.g. the frontend metadata dataclasses serialized by Pydantic, where a :class:LazyString cannot go). The stored source string is translated later by passing it through :func:gettext at the rendering boundary. Extraction picks the literals up via pybabel extract -k gettext_noop.

lazy_gettext

lazy_gettext(message: str) -> LazyString

Mark message for translation but defer it to render time.

For module-level constants and other values built before a request locale exists. Extraction picks these up via pybabel extract -k lazy_gettext.

Encryption

sparkth.lib.encryption

Public API for the symmetric-encryption service.

Import get_encryption_service (and EncryptionService) from here instead of reaching into sparkth.core.encryption directly. It encrypts stored secrets — such as LLM API keys — at rest with Fernet. Implementation lives in sparkth/core/encryption.py.

Data models

sparkth.lib.models

Public API for the SQLModel data models that plugins consume.

Plugins import model classes and the shared mixins from here instead of reaching into sparkth.core.models.* directly — every internal symbol a plugin imports becomes an implicit public API and blocks refactoring (see issue #379).

Implementation lives in sparkth/core/models/.

TZDateTime

Bases: TypeDecorator[datetime]

Column type for datetimes that are timezone-aware UTC on both sides.

PostgreSQL stores timestamptz natively, but SQLite drops the UTC offset on storage, so values that read back naive are re-tagged as UTC. Naive datetimes are rejected on write — callers must store aware values (see :func:utc_now).

utc_now

utc_now() -> datetime

Return the current UTC datetime.

Authentication

sparkth.lib.auth

Bearer-token authentication: the token-reading helpers and the current-user dependency.

The single canonical home for get_current_user: every caller (routes, the permission gate, plugins, and the test harness) imports it from here. It is deliberately not re-exported from sparkth.api.v1.auth — one object keeps FastAPI dependency overrides working.

get_current_user is a FastAPI dependency, so it is only available to code that runs inside a route. Code that must identify the caller earlier — PluginAccessMiddleware runs before routing, so no dependency has resolved yet — composes the same two helpers this module builds the dependency from, :func:decode_token_username and :func:get_user_by_username, rather than decoding tokens or querying users of its own. One implementation of "who is this request from" keeps the security gate from drifting away from the dependency as tokens or user lookup change.

bind_interface_locale

bind_interface_locale(user: User) -> None

Install user's stored language as the request's interface locale.

Called from both of get_current_user's exits, and from PluginAccessMiddleware. The cached path matters as much as the full lookup: the gate populates request.state.user on every authenticated plugin route, so binding only on the lookup path would leave the stored preference unapplied on all of them. The gate calls this itself as well, because the 403 it renders never reaches a route for the dependency to run — by then it has decoded the token and loaded the user, so the preference is known.

The membership check is what keeps a language withdrawn from the allowlist from being honoured. resolve_language is deliberately not used — its fallback would overwrite a negotiated Accept-Language with DEFAULT_LANGUAGE, which is worse than leaving the header in charge.

decode_token_username

decode_token_username(token: str) -> str | None

Return the username a bearer token identifies, or None if it identifies nobody.

None covers every way a token can fail to name a user — malformed, signed with the wrong key, expired, or carrying no sub claim — because callers treat them alike: an unreadable token is an unauthenticated request. Callers decide what that means; this helper never raises.

Parameters:

Name	Type	Description	Default
token	str	The raw JWT from the Authorization header, without the Bearer prefix.	required

get_current_user async

get_current_user(
    request: Request,
    credentials: HTTPAuthorizationCredentials = Depends(
        security_scheme
    ),
    session: AsyncSession = Depends(get_async_session),
) -> User

Resolve the authenticated user, rejecting the request when the token names no one.

Reuses the user PluginAccessMiddleware left on request.state when it has already resolved this request's caller, rather than decoding the same token and re-reading the same row. Only that gate writes the attribute, and only from this request's own token, so the answer is the one this dependency would have computed. Requests it never identified — core routes, which it does not gate — fall through to the full lookup.

The reused instance is detached: the gate's session has closed by the time the route runs. That is safe because User maps only columns, which stay readable on a detached instance; a test in tests/core/plugins/test_middleware.py pins that, since adding a relationship to User is what would make this unsafe.

A resolved user's stored language replaces the locale the middleware negotiated from Accept-Language, so translated copy follows what the user chose rather than what their browser advertises, on both exits below — the cached one and the full lookup. An unset or unsupported stored value leaves the negotiated locale in place.

get_user_by_username async

get_user_by_username(
    username: str, session: AsyncSession
) -> User | None

Return the user with this username, or None when no user has it.

RAG

sparkth.lib.rag

Public API for the RAG library.

All plugins and external modules import RAG functionality from here.

DocumentNotFoundError

Bases: RAGError

Raised when a document does not exist or is not accessible to the user.

DocumentSection dataclass

DocumentSection(
    source_name: str,
    chapter: str | None,
    section: str | None,
    subsection: str | None,
    chunk_count: int,
    position_index: int,
)

A structural section in a document, with chunk count and document order.

__repr__

__repr__() -> str

Match the previous Pydantic-style repr used in RAG agent tool output.

__str__

__str__() -> str

Use the stable agent-facing representation for string conversion.

RAGNotReadyError

RAGNotReadyError(document_id: int, status: str)

Bases: RAGError

Raised when the document exists but status is not READY.

RAGRetrievalError

Bases: RAGError

Raised when agent retrieval or section-chunk fetch fails.

RetrievedChunk dataclass

RetrievedChunk(
    source_name: str,
    chapter: str | None,
    section: str | None,
    subsection: str | None,
    content: str,
)

A chunk returned by retrieval, with its document/section attribution.

ScannedPDFError

ScannedPDFError(source_name: str)

Bases: RAGError

Raised when a PDF appears to be scanned/image-only.

UnsupportedFileTypeError

Bases: RAGIngestionError

Raised when a file's type cannot be handled by the RAG extractors.

Callers typically treat this as a benign skip (the file is fine, it just isn't RAG-ingestible).

agentic_retrieve_context async

agentic_retrieve_context(
    query: str, document_ids: list[int], llm: BaseChatModel
) -> list[RetrievedChunk]

Retrieve relevant document chunks for a query across the given documents.

Validates that every document exists and is READY before retrieval — callers do not need to perform this check themselves. Uses agentic section retrieval per document and returns a flat list of RetrievedChunk. Opens its own database sessions.

Parameters:

Name	Type	Description	Default
document_ids	list[int]	Documents to search. All must exist and be READY.	required
query	str	The user's natural-language query.	required
llm	BaseChatModel	LangChain chat model used by the retrieval agent.	required

Returns:

Type	Description
list[RetrievedChunk]	Flat list of RetrievedChunk across all documents (empty if no matches).

Raises:

Type	Description
DocumentNotFoundError	a document is missing or soft-deleted.
RAGNotReadyError	a document exists but is not READY.
RAGRetrievalError	retrieval failed.

copy_document_chunk_links async

copy_document_chunk_links(
    session: AsyncSession,
    source_document_id: int,
    target_document_id: int,
) -> None

Copy all chunk links from source_document_id to target_document_id.

format_document_chunks_as_llm_context

format_document_chunks_as_llm_context(
    retrieved_chunks: list[RetrievedChunk],
) -> str

Group retrieved chunks by document and render them as one LLM context string.

Retrieval can return chunks from several documents interleaved in relevance order. They are grouped by the document each chunk came from (keyed by the document name the chunk carries, in first-seen order); each document's chunks are rendered as a labelled context block, and the blocks are joined with a blank line into a single string. An empty input yields an empty string.

get_rag_ingested_document_structure async

get_rag_ingested_document_structure(
    document_id: int,
) -> list[DocumentSection]

Return ordered section metadata generated from the ingested RAG chunks.

Sections are ordered by the minimum chunk id within each (chapter, section, subsection) group. This preserves document insertion order, so position_index reliably reflects the section's position in the original document.

Returns an empty list when the document does not exist.

ingest_document async

ingest_document(
    filename: str, file_bytes: bytes, document_id: int
) -> IngestionResult

Ingest a document's bytes into the RAG store.

Pipeline: eligibility check -> extract -> chunk -> store (with cross-document content-hash dedup) -> link chunks to document_id. Opens and commits its own database session.

Audited as a rag.document_ingested event. A success is recorded in the same transaction as the chunk write (fail-closed, so unrecordable content cannot enter the corpus). A failure is recorded, in its own transaction before the error propagates, for the two eligibility/extraction errors this function declares below; any other error (a parse failure inside an extractor, a chunker error, a database error from the chunk write) propagates without an audit record. Nothing entered the corpus in those cases, so the trail stays accurate about the corpus itself, but it is not a complete log of attempts.

Parameters:

Name	Type	Description	Default
document_id	int	Document.id recorded in the chunk-link table.	required
file_bytes	bytes	Raw file content.	required
filename	str	Original filename (drives extension dispatch).	required

Returns:

Type	Description
IngestionResult	IngestionResult with new/reused chunk counts.

Raises:

Type	Description
UnsupportedFileTypeError	type the extractors cannot handle.
ScannedPDFError	PDF appears scanned/image-only.

LLM

sparkth.lib.llm

Public API for the LLM library.

All plugins and external modules import LLM functionality from here. Nothing outside sparkth/llm/ should import from sparkth.llm.* directly.

BaseChatProvider

BaseChatProvider(
    api_key: str,
    model: str,
    system_prompt: str | None = None,
    temperature: float = 0.7,
    max_tool_executions: int = 50,
    max_retries: int = 2,
)

Bases: ABC

create_llm

create_llm(
    streaming: bool = False,
    callbacks: list[Any] | None = None,
) -> Any

Public interface for creating a configured LangChain LLM instance.

send_message async

send_message(
    messages: list[dict[str, Any]],
    max_tokens: int | None = None,
    tools: list[Any] | None = None,
) -> dict[str, Any]

Send a message and get a response, with optional tool usage.

stream_message async

stream_message(
    messages: list[dict[str, Any]],
    max_tokens: int | None = None,
    tools: list[Any] | None = None,
) -> AsyncIterator[dict[str, Any]]

Stream a message response, with optional tool usage.

Yields typed event dicts

{"type": "token", "content": str} — text to display {"type": "tool_start","name": str} — tool execution beginning {"type": "tool_end", "name": str} — tool execution finished

LLMConfigAdapter

Base adapter for plugins that hold an optional llm_config_id reference.

preprocess_config: validates the referenced LLMConfig is owned by the user. postprocess_config: resolves llm_config_id to name/provider/model for the frontend.

LLMConfigDuplicateNameError

LLMConfigDuplicateNameError(name: str)

Bases: ValueError

Raised when an LLM config with the same name already exists for the user.

LLMConfigInactiveError

LLMConfigInactiveError()

Bases: ValueError

Raised when an LLM config exists but is inactive.

LLMConfigModelNotSetError

Bases: ValueError

Raised when the model field is empty on an LLM config.

LLMConfigNotFoundError

LLMConfigNotFoundError(config_id: int, user_id: int)

Bases: ValueError

Raised when an LLM config is not found.

LLMConfigService

LLMConfigService(
    encryption: SupportsEncryption, cache: SupportsCache
)

Initialize with encryption and cache services.

create async

create(
    session: AsyncSession,
    user_id: int,
    name: str,
    provider: str,
    model: str,
    api_key: str,
) -> LLMConfig

Create a new LLM config, encrypting the API key. Raises LLMConfigDuplicateNameError on duplicate name.

delete async

delete(
    session: AsyncSession, user_id: int, config_id: int
) -> bool

Soft-delete a config and evict its cache entry. Returns False if not found.

get async

get(
    session: AsyncSession, user_id: int, config_id: int
) -> LLMConfig | None

Fetch a single non-deleted LLM config by ID and user, or None if not found.

list async

list(
    session: AsyncSession,
    user_id: int,
    include_inactive: bool = False,
) -> list[LLMConfig]

Return all non-deleted LLM configs for a user, newest first.

By default only active configs are returned. Pass include_inactive=True to also include deactivated configs (e.g. for the settings page).

mask_key staticmethod

mask_key(api_key: str) -> str

Return a masked version of the API key, preserving the provider prefix and last 4 chars.

resolve async

resolve(
    session: AsyncSession, user_id: int, config_id: int
) -> tuple[LLMConfig, str]

Return (config, decrypted_api_key). Caches by (user_id, config_id). Updates last_used_at.

rotate_key async

rotate_key(
    session: AsyncSession,
    user_id: int,
    config_id: int,
    api_key: str,
) -> LLMConfig

Replace the API key for a config, updating encryption and invalidating cache.

set_active async

set_active(
    session: AsyncSession,
    user_id: int,
    config_id: int,
    is_active: bool,
) -> LLMConfig

Activate or deactivate a config. Bypasses the is_active filter so inactive configs can be re-activated.

update async

update(
    session: AsyncSession,
    user_id: int,
    config_id: int,
    name: str | None = None,
    model: str | None = None,
) -> LLMConfig

Update name and/or model of an existing config. Raises ValueError if not found or duplicate name.

LLMConfigValidationError

Bases: ValueError

Raised when a field value is invalid (e.g. model not allowed for provider).

get_provider

get_provider(
    provider_name: str,
    api_key: str,
    model: str,
    system_prompt: str | None = None,
    temperature: float = 0.7,
    max_tool_executions: int = 50,
    max_retries: int = 2,
) -> BaseChatProvider

Get a chat provider instance.

get_provider_catalog

get_provider_catalog() -> list[ProviderCatalogEntry]

Return all providers with their available models.

Analytics

sparkth.lib.analytics

Public API for the analytics emission gateway.

All application code and plugins import analytics functionality from here rather than reaching into sparkth.core.analytics.* directly. Implementation lives in sparkth/core/analytics/.

Plugins

• subclass AnalyticsEventSchema to define an event payload schema (declaring event_type/version),
• register it from their __init__ via register_event_schema(self, MyEvent),
• emit it through emit_event (reach for the lower-level ingest_event only when the caller already holds an analytics session).

AnalyticsEventSchema

Bases: BaseModel

Base class for analytics event payload schemas.

Subclasses set event_type and version as class attributes and declare the payload as ordinary Pydantic fields. Those two are ClassVar — identity metadata, not part of the validated payload — and are required: a subclass that omits either fails at definition time (__init_subclass__).

Extra fields are forbidden so a producer sending unexpected keys gets a 422 rather than having those fields silently dropped from the stored row.

ContinuousAggregateNotFound

ContinuousAggregateNotFound(name: str)

Bases: Exception

Raised when a backfill targets a continuous aggregate that does not exist.

DuplicateEventTypeError

DuplicateEventTypeError(event_type: str, version: int)

Bases: Exception

Raised when a schema claims an already-registered (event_type, version).

Fires only for a different class claiming the identity — a startup-fatal programming error, because a producer's payload could then silently validate against the wrong schema. Re-registering the same class is a no-op, so a plugin constructed more than once does not trip this.

EventNamespaceError

EventNamespaceError(plugin_name: str, event_type: str)

Bases: Exception

Raised when a plugin contributes an event not namespaced under its own name.

LoginActivityPoint

Bases: BaseModel

One day's login count. day is an ISO YYYY-MM-DD string.

UnknownEventTypeError

UnknownEventTypeError(event_type: str, version: int)

Bases: Exception

Raised when an (event_type, version) pair has no registered schema.

backfill_continuous_aggregates async

backfill_continuous_aggregates(
    name: str | None = None,
) -> list[str] | None

Materialize the full history of continuous aggregates (all, or one by name).

Continuous aggregates are created WITH NO DATA (so creation does not backfill inside Alembic's transaction), and their refresh policies only cover a trailing window (start_offset). Without a one-off full refresh, buckets older than that window fall below the materialization watermark once the first policy run advances it and disappear from the view — so any pre-migration history is silently lost. Run this once after applying an aggregate's migration on PostgreSQL/TimescaleDB; it is idempotent and safe to re-run.

Parameters:

Name	Type	Description	Default
name	str | None	Refresh only this aggregate. When None (default), refresh every continuous aggregate discovered in the TimescaleDB catalog.	None

Returns:

Type	Description
list[str] | None	The list of aggregate names refreshed (possibly empty if none are registered), or
list[str] | None	None if skipped because the analytics database is not PostgreSQL/TimescaleDB
list[str] | None	(e.g. SQLite in tests/e2e, where continuous aggregates do not exist and the read
list[str] | None	path aggregates raw_events directly).

Raises:

Type	Description
ContinuousAggregateNotFound	if name is given but no such aggregate exists.

emit_event async

emit_event(
    event_type: str,
    version: int,
    payload: dict[str, Any],
    actor_id: str | None = None,
) -> None

Validate and land an analytics event, propagating any failure.

The producer-facing counterpart to :func:ingest_event: it opens its own analytics session, so callers need no session plumbing. That is the only thing it adds. It catches nothing — UnknownEventTypeError, ValidationError, SQLAlchemyError and anything else reach the caller unchanged, so a broken analytics write is never silently hidden.

Producers call this from FastAPI background tasks and detached asyncio tasks, which run after the response has been sent; a failure there surfaces as an unhandled task error in the logs rather than affecting the request that was being measured.

Prefer ingest_event where the caller already holds an analytics session.

Parameters:

Name	Type	Description	Default
event_type	str	The registered event name, e.g. "chat.message_sent".	required
version	int	The schema version, e.g. 1.	required
payload	dict[str, Any]	The event body, validated against the registered schema.	required
actor_id	str | None	The acting user's id as a string, stored for provenance.	None

Raises:

Type	Description
UnknownEventTypeError	No schema is registered for this type and version.
ValidationError	The payload does not match the registered schema.
SQLAlchemyError	The analytics database could not be reached or written.

get_event_schema

get_event_schema(
    event_type: str, version: int
) -> type[AnalyticsEventSchema]

Return the schema for (event_type, version), or raise UnknownEventTypeError.

get_login_activity async

get_login_activity(
    session: AsyncSession, days: int = 30
) -> list[LoginActivityPoint]

Return daily login counts, newest first, for the last days calendar days.

The window is bounded by a date floor (now - days), so gap days never let the series reach back beyond the requested window. Days with no logins are omitted entirely — the series is not zero-filled; callers must tolerate gaps.

ingest_event async

ingest_event(
    session: AsyncSession,
    event_type: str,
    version: int,
    payload: dict[str, Any],
    actor_id: str | None = None,
    occurred_at: datetime | None = None,
) -> None

Validate payload against the registered schema and land it in raw_events.

Parameters:

Name	Type	Description	Default
session	AsyncSession	An async session bound to the analytics database.	required
event_type	str	The base event name, e.g. "assessment.submitted".	required
version	int	The schema version, e.g. 1.	required
payload	dict[str, Any]	The raw event body to validate.	required
actor_id	str | None	The authenticated user id, stored for provenance.	None
occurred_at	datetime | None	When the event happened; defaults to now(UTC).	None

Raises:

Type	Description
UnknownEventTypeError	No schema is registered for (event_type, version).
ValidationError	The payload does not satisfy the schema.
SQLAlchemyError	The insert failed.

register_event_schema

register_event_schema(
    plugin: SparkthPlugin,
    schema: type[AnalyticsEventSchema],
) -> None

Register a plugin's event schema on the ANALYTICS_EVENTS hook.

Call this from a plugin's __init__. Registration happens at import time, straight into the ANALYTICS_EVENTS hook the gateway resolves against.

Two startup-fatal guards, enforced here so a misconfigured plugin crashes the process at import rather than at first emit (a third — that the schema declares event_type/version — is enforced on AnalyticsEventSchema itself, at class-definition time, via __init_subclass__):

• Namespace. event_type must be prefixed with the contributing plugin's name (e.g. plugin slack → "slack.*"), else EventNamespaceError. This stops a plugin squatting a core or another plugin's event name.
• Collision. A different class claiming an already-registered (event_type, version) raises DuplicateEventTypeError. Re-registering the identical class is a no-op, so constructing a plugin more than once — a module re-import, or a test building its own instance — is not a collision. This follows KeyedClassHook's rule for class registries; only a different class can squat a name, which is what the guard is for.

HTTP client

sparkth.lib.http

BaseHttpClient

BaseHttpClient(base_url: str, auth: Auth = Auth.BEARER)

Shared base for HTTP API clients that authenticate with a token.

Subclasses provide a token property and call _request() instead of duplicating the token-check + URL-join logic. The optional base_url override on _request supports clients whose verb methods accept a per-call base URL (e.g. OpenEdxClient).

Initialise the client with a base URL and authentication scheme.

token property

token: str | None

Return the bearer token for the current session; override in subclasses.

__aenter__ async

__aenter__() -> Self

Return self to support use as an async context manager.

__aexit__ async

__aexit__(
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: TracebackType | None,
) -> None

Close the underlying session on context manager exit.

close async

close() -> None

Close the underlying aiohttp session if it is still open.

Documents

sparkth.lib.documents

Public API for the Document registry.

All plugins and external modules manage document identity and status from here.

Document

Bases: TimestampedModel, SoftDeleteModel

Plugin-agnostic registry of documents submitted for RAG ingestion.

Plugins create one row per source document they want to ingest. RAG uses document_id as its primary reference — it never imports from any plugin model.

create_document async

create_document(
    session: AsyncSession,
    user_id: int,
    name: str,
    mime_type: Optional[str],
) -> Document

Create a new Document row with QUEUED status and flush (no commit).

get_document async

get_document(
    session: AsyncSession, document_id: int, user_id: int
) -> Optional[Document]

Return the Document if owned by user_id and not soft-deleted; else None.

list_ready_documents async

list_ready_documents(
    session: AsyncSession, user_id: int
) -> list[Document]

Return non-deleted READY documents owned by user_id, ordered for display.

soft_delete_document async

soft_delete_document(
    session: AsyncSession, document_id: int
) -> None

Soft-delete a Document by id. Does not commit.

Records a rag.document_deleted audit event in the caller's transaction, so the deletion and its record commit or roll back together. A missing id is a no-op and records nothing.

update_document_status async

update_document_status(
    session: AsyncSession,
    document_id: int,
    status: DocumentStatus,
    error: Optional[str] = None,
) -> None

Update status (and optional error) on an existing Document. Does not commit.

Enums

sparkth.lib.enums

Exceptions

sparkth.lib.exceptions

Legacy HTTP-carrying exceptions, kept for existing LMS-client callers.

These classes predate the exception→HTTP mapping standard (CLAUDE.md, "Domain exceptions → HTTP responses"): they carry status_code on the exception itself. New code must NOT follow this pattern — raise an HTTP-agnostic domain exception and map it to a status via register_exception_handler (sparkth.lib.exceptions.handlers) instead.