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.

Example

from sparkth.lib.settings import get_settings

settings = get_settings()
print(settings.SECRET_KEY)

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/.

utc_now

utc_now() -> datetime

Return the current UTC datetime.

Authentication

sparkth.lib.auth

Authentication dependency for resolving the current user from a bearer token.

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.

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.

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 ingest_event.

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).

Registration is not idempotent: this fires for a colliding different class and for re-registering the same class. Either is a startup-fatal programming error — a producer's payload could silently validate against the wrong schema.

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.

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. Any class claiming an already-registered (event_type, version) raises DuplicateEventTypeError.

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.

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.