API Reference

This page brings together the project modules that make up the public backend surface. Start with the short narratives for an overview of each component and then dive into the autogenerated member listings for the complete API.

Working With This Reference 

Prefer the summaries for intent and integration tips, then jump to the generated API listings when you need argument details.
Environment variables called out below match the defaults baked into the codebase; override them per deployment via the usual .env or container settings.
All modules follow FastAPI best practices and integrate cleanly with the application factory shown later. Import paths in examples assume you are running from inside the backend package root.

Application Lifecycle and HTTP Surface 

`app_factory`

Creates the FastAPI application, wires every router, and enforces the Config UI login flow. Highlights:

Session cookies are signed (HMAC) and paired with CSRF tokens; defaults are safe for HTTPS deployments and configurable for local testing.
Rate limiting uses an in-process counter for the Config UI.
Static assets for the Config UI are served from assets/, keeping the backend self-contained for local runs.

A minimal entry point is:

from fastapi import FastAPI
from app_factory import create_app

app: FastAPI = create_app()

Key environment variables
Variable	Purpose	Default
`CFG_UI_SECRET`	HMAC key for Config UI sessions/CSRF (falls back to `JWT_SECRET`).	`replace-me`
`CFG_UI_TTL_MIN`	Session lifetime in minutes for the Config UI.	`30`
`CFG_UI_PASSWORD` / `CFG_UI_PASSWORD_FILE`	Plaintext Config UI password, hashed on startup.	unset
`SECURE_COOKIES`	Toggle the `Secure` cookie flag for local HTTP testing.	`true`
`CFG_UI_RL_*`	Sliding window and lock thresholds for login throttling.	see module defaults

`lifespan`

Declares the async lifespan handler used by FastAPI to stage supporting services during startup and shutdown. It:

Initialises the in-memory pubsub backend through pubsub.init_pubsub().
Ensures the widget registry is ready — creating indexes and optionally pre-warming bundles.
Starts configured producer tasks from endpoint_config.get_stream_funcs() and cancels them cleanly when the server stops.

lifespan.lifespan(app)[source]

Authentication and Session Management 

`auth`

Handles OAuth2-style login for API clients. Important aspects:

Local users are stored in-memory with bcrypt hashes; MSAL users are auto-provisioned without stored passwords and always carry the SKAO group.
Access tokens are JSON Web Tokens (HS256) carrying role/group claims. The module guards endpoints via FastAPI dependencies to enforce authentication.
Token TTL is intentionally short by default to keep local testing snappy; set TOKEN_TTL_MINUTES to lengthen sessions in real deployments.

Authentication environment variables
Variable	Purpose	Default
`TOKEN_TTL_MINUTES`	Lifetime for every issued access token.	`1`
`JWT_SECRET`	Symmetric signing key shared with the Config UI session code.	`Replace me with a random string!`

exception auth.HTTPException(status_code, detail=None, headers=None)[source]

Bases: HTTPException

An HTTP exception you can raise in your own code to show errors to the client.

This is for client errors, invalid authentication, invalid data, etc. Not for server errors in your code.

Read more about it in the [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/).

## Example

```python from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {“foo”: “The Foo Wrestlers”}

@app.get(“/items/{item_id}”) async def read_item(item_id: str):

if item_id not in items:
raise HTTPException(status_code=404, detail=”Item not found”)

return {“item”: items[item_id]}

```

Parameters:

status_code (Annotated[int, Doc('\n HTTP status code to send to the client.\n\n Read more about it in the\n [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#use-httpexception)\n ')])
detail (Annotated[Any, Doc('\n Any data to be sent to the client in the `detail` key of the JSON\n response.\n\n Read more about it in the\n [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#use-httpexception)\n ')])
headers (Annotated[Mapping[str, str] | None, Doc('\n Any headers to send to the client in the response.\n\n Read more about it in the\n [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#add-custom-headers)\n\n ')])

Return type:

None

class auth.Token(*, access_token, token_type='bearer', expires_in)[source]

Bases: BaseModel

Parameters:

access_token (str)
token_type (str)
expires_in (int)

access_token: str

token_type: str

expires_in: int

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class auth.User(*, username, full_name=None, role='Viewer', role_level=1, groups=<factory>)[source]

Bases: BaseModel

Parameters:

username (str)
full_name (str | None)
role (str)
role_level (int)
groups (list[str])

username: str

full_name: str | None

role: str

role_level: int

groups: list[str]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class auth.MsalPayload(*, id_token)[source]

Bases: BaseModel

Parameters:: id_token (str)

id_token: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

auth.canonical_role(role)[source]

Parameters:: role (str | None)
Return type:: str

auth.role_level(role)[source]

Parameters:: role (str | None)
Return type:: int

auth.has_required_role_level(role, minimum)[source]

Parameters:

role (str | None)
minimum (int)

Return type:

bool

async auth.get_current_user(request, token=Depends(dependency=<fastapi.security.oauth2.OAuth2PasswordBearer object>, use_cache=True, scope=None), query_token=Query(None))[source]

Shared dependency - accepts header / query / cookie.

Parameters:

request (Request)
token (str | None)
query_token (str | None)

Return type:

User

auth.sync_env_users()[source]

Return type:: None

async auth.login(response, form=Depends(dependency=None, use_cache=True, scope=None))[source]

Parameters:

response (Response)
form (OAuth2PasswordRequestForm)

Return type:

Token

async auth.msal_login(payload, response)[source]

The frontend calls this route after MSAL acquires an ID-token.

We trust any token MSAL gives us for local dev - no signature check. Add proper validation later (see suggestions below).
We decode the token, map claims → username/full_name, provision a local user if necessary, then hand out a normal taranta_jwt + cookie.

Parameters:

payload (MsalPayload)
response (Response)

Return type:

Token

async auth.logout(response)[source]

Parameters:: response (Response)

async auth.read_me(current=Depends(dependency=<function get_current_user>, use_cache=True, scope=None))[source]

Parameters:: current (User)

async auth.validate_token(request, token=Depends(dependency=<fastapi.security.oauth2.OAuth2PasswordBearer object>, use_cache=True, scope=None), query_token=Query(None))[source]

Quick session check - handy from tools like curl or the browser:

GET /auth/validate?token=<access_token>

or GET with Authorization: Bearer … or GET with the cookie already set

Parameters:

request (Request)
token (str | None)
query_token (str | None)

async auth.list_roles_and_groups()[source]: Return roles and groups from the in-memory user store for UI dropdowns.

Configuration Management 

`config_editor`

Exposes the Config UI endpoints to read, validate, diff, and activate stored endpoint configuration documents. Every mutating route enforces the trio of a valid Config UI session, CSRF token, and same-origin request to defend against browser threats. Payloads are validated through utils.validation before they are persisted or activated.

async config_editor.get_config(version=Query(None))[source]

Return the requested version (or active) plus a varsResolved field.

Parameters:: version (str | None)

async config_editor.get_schema()[source]

async config_editor.get_versions()[source]: List all saved versions, the active version, the currently loaded version, and whether the loader runs in recovery mode.

async config_editor.activate_version(version)[source]

Activate an existing version.

Parameters:: version (str)

async config_editor.delete_version(version)[source]

Permanently delete a non-active version from the store.

Parameters:: version (str)

async config_editor.save_config(payload)[source]

Validate and save a new version to MongoDB, returning the diff against the previous active version, and trigger a graceful reload when activate is true.

Request shape (both supported):

Back-compat: raw endpoints.json object (auto-bumps patch & activates)
New: {
“config”: <endpoints.json>, “version”: “1.2.4”, # optional, defaults to auto-bump patch “activate”: true, # optional, defaults to true “message”: “why” # optional

}

Parameters:: payload (dict[str, Any])

async config_editor.diff_payload(payload=Body(PydanticUndefined))[source]

Compute a unified diff between two payloads:

base: requested version (when provided) or the active version
new : the submitted payload (raw endpoints JSON)

Request body accepts either:

{ “config”: <json>, “baseVersion”: “1.2.3” } # preferred <json> # legacy shape (no baseVersion)

Returns: { ok: true, diff: “<unified diff>” }

Parameters:: payload (dict[str, Any])

async config_editor.diff_versions(frm=Query(None), to=Query(None))[source]

Compute a unified diff between two SAVED versions.

Query params:: ?from=1.2.3&to=1.2.4

If ‘to’ is omitted, compares against the active version. If ‘from’ is omitted, compares active (old) with ‘to’ (new).

Returns: { ok: true, diff: “<unified diff>” }

Parameters:

frm (str | None)
to (str | None)

async config_editor.test_local(func_name, payload=Body({}))[source]

Execute a local function declared in endpoints.json; return raw output.

Parameters:

func_name (str)
payload (dict[str, Any])

`config_store`

Wraps MongoDB collections that persist versioned endpoint configuration state. The helpers here are intentionally blocking because they run in worker threads under FastAPI. Use them to save new versions, promote a version to “active”, read history, or bootstrap the database from endpoint_config/endpoints.json.

Configuration storage variables
Variable	Purpose	Default
`MONGO_URI`	Connection string for MongoDB (shared with other stores).	`mongodb://mongo:27017`
`MONGO_DB`	Database name that houses configuration and preferences.	`octopus`
`CONFIG_COLLECTION`	Collection storing each saved endpoint configuration.	`endpoint_configs`
`CONFIG_STATE_COLLECTION`	Stores current/last-known-good version metadata.	`endpoint_config_state`

config_store.get_active_version()[source]

Return the currently active version stored in state (or None).

Return type:: str | None

config_store.get_last_good_version()[source]

Return the last-known-good version (if any).

Return type:: str | None

config_store.set_last_good_version(version)[source]

Persist last-known-good version (best-effort).

Parameters:: version (str)
Return type:: None

config_store.list_versions()[source]

Return type:: dict[str, Any]

config_store.fetch_active_config()[source]

Return type:: dict[str, Any] | None

config_store.fetch_config_by_version(version)[source]

Parameters:: version (str | None)
Return type:: dict[str, Any] | None

config_store.health_check()[source]

Lightweight connectivity check. Raises on failure so callers can differentiate config-not-found from Mongo-unreachable cases.

Return type:: None

config_store.set_active_version(version)[source]

Parameters:: version (str)
Return type:: None

config_store.save_new_version(config, *, version, activate, message=None)[source]

Insert a new version document. When version is None, bump the patch from the active version (or the latest, or 1.0.0 if none exist).

Parameters:

config (dict[str, Any])
version (str | None)
activate (bool)
message (str | None)

Return type:

str

config_store.ensure_bootstrap_from_file(cfg_path)[source]

Load endpoints.json and store it as version 1.0.0 (active) when the collection is empty.

Parameters:: cfg_path (Path)
Return type:: None

config_store.delete_version(version)[source]

Delete a saved config version (refuse when it is the active version) and adjust the last_good pointer when needed.

Parameters:: version (str)
Return type:: None

`preferences_store`

Provides the async MongoDB-backed persistence for user-specific UI preferences. It is intentionally lightweight — fetch and update helpers are all you need for widgets arranging their own state.

Storage is routed through the same MONGO_URI/MONGO_DB as the config store; override the collection with MONGO_COLLECTION (defaults to preferences).
fetch_preferences also assembles a shared_workspace_library from every user’s published shared_workspace_grants, attaching the latest owner workspace snapshot and variables so shared dashboards stay in sync without a manual export. Persisted shared_workspace_library data is ignored and rebuilt on every fetch to avoid stale shared copies.
update_preferences now applies partial JSON patches on top of the stored document instead of replacing the entire config. This keeps workspace saves small and prevents unrelated settings from being lost when only one field changes.
Role and group picklists for the share dialog come from the /auth/roles helper endpoint, which returns the canonical Admin/Viewer values plus any extras present in the in-memory user store.

Shared workspaces and the Viewer role

Owners can grant access to a workspace by role, group, or explicit username. Each grant also carries an access mode: view or edit.
The frontend filters grants against the caller’s JWT claims before rendering. The backend now also authorizes collaborative shared-workspace updates when a non-owner sends a workspaceTarget patch for a workspace they were granted with access=edit.
Incoming preference updates are normalised to snake_case (shared_workspace_grants) for predictable querying. The owner continues to store the authoritative copy of the workspace. Shared recipients only write back to the owner document for explicit editable shared-workspace mutations; everything else stays in the recipient profile. Full workspaces replacement patches are excluded from that collaborative path so shared editors cannot rename or delete the owner workspace by sending a broad replacement payload.
After every successful preference mutation, the resolver publishes a preferences-updated invalidation event through GraphQL pubsub. Frontend clients subscribe to that channel and refetch preferences so owner and recipient sessions converge on the stored workspace state.
Viewer-role users are treated as view-only by the frontend before data reaches MongoDB: only currentWorkspace, theme, and timeRange keys are persisted and any attempts to mutate widgets, panels, or variables are dropped.

async preferences_store.can_edit_shared_workspace(owner_id, username, workspace_name, role=None, groups=None)[source]

Parameters:

owner_id (str)
username (str)
workspace_name (str)
role (str | None)
groups (list[str] | None)

Return type:

bool

async preferences_store.fetch_preferences(user_id)[source]

Parameters:: user_id (str)
Return type:: dict[str, Any]

async preferences_store.fetch_workspace_names(user_id)[source]

Parameters:: user_id (str)
Return type:: list[str]

async preferences_store.fetch_workspace_snapshot(user_id, workspace_name)[source]

Parameters:

user_id (str)
workspace_name (str)

Return type:

dict[str, Any] | None

async preferences_store.fetch_preferences_summary(user_id)[source]

Parameters:: user_id (str)
Return type:: dict[str, Any]

async preferences_store.set_current_workspace(user_id, workspace_name)[source]

Parameters:

user_id (str)
workspace_name (str)

Return type:

dict[str, Any]

async preferences_store.save_workspace(user_id, workspace_name, workspace, make_current=False)[source]

Parameters:

user_id (str)
workspace_name (str)
workspace (dict[str, Any])
make_current (bool)

Return type:

dict[str, Any]

async preferences_store.delete_workspace(user_id, workspace_name, next_workspace=None)[source]

Parameters:

user_id (str)
workspace_name (str)
next_workspace (str | None)

Return type:

dict[str, Any]

async preferences_store.rename_workspace(user_id, old_name, new_name)[source]

Parameters:

user_id (str)
old_name (str)
new_name (str)

Return type:

dict[str, Any]

async preferences_store.update_preferences(user_id, config)[source]

Parameters:

user_id (str)
config (dict[str, Any])

Return type:

dict[str, Any]

Messaging and Streaming 

`pubsub`

Offers an in-process publish/subscribe helper layer. Use init_pubsub() during startup, then:

Call publish() or run_producer() to push JSON serialised messages to channels.
Consume updates through subscribe() which yields decoded JSON payloads.
Messages stay in-process so application behaviour remains deterministic.

async pubsub.init_pubsub()[source]

Create an in-memory pubsub backend. This is process-local only.

Return type:: None

async pubsub.publish(channel, message)[source]

JSON-encode message and push it to channel.

Parameters:

channel (str)
message (Any)

Return type:

None

async pubsub.run_producer(channel, producer, poll_interval=1.0)[source]

Continuously feed a producer into the in-memory pubsub backend.

async-generator functions - every yielded item is published.
coroutine functions - called every poll_interval s, their returned value is published.

Parameters:

channel (str)
producer (Callable[[...], Any])
poll_interval (float)

Return type:

None

async pubsub.subscribe(channel_name)[source]

Yield decoded JSON messages arriving on channel_name.

Parameters:: channel_name (str)
Return type:: AsyncIterator[Any]

Widgets Registry 

`widgets_registry`

Manages the lifecycle of embeddable UI widget bundles. Capabilities include:

Downloading, integrity-checking, and caching widget bundles from approved CDNs or artefact repositories.
Enforcing organisation-scoped npm package naming to prevent arbitrary third party code execution.
Maintaining an LRU in-memory cache plus MongoDB metadata so frequently-used bundles stay warm even across restarts.
Providing startup helpers (ensure_ready) and graceful shutdown of the shared aiohttp client session.

Widget registry variables
Variable	Purpose	Default
`WIDGET_COLLECTION`	MongoDB collection tracking widget bundle metadata.	`widget_versions`
`WIDGET_CACHE_DIR`	Filesystem cache for extracted bundles.	`widgets-cache`
`WIDGET_ALLOWED_HOSTS`	Comma-separated whitelist of hosts allowed for downloads.	`cdn.jsdelivr.net`, `unpkg.com`, `artefact.skao.int`
`WIDGET_ALLOWED_NPM_SCOPES`	Scoped npm packages permitted for install.	`@ska-octopus-widgets`
`WIDGET_PREWARM`	Whether to pre-fetch allowed bundles during startup.	`true`