Python API reference¶

Generated from docstrings with Sphinx autodoc. Public modules:

FluxLit public API.

FluxLit unifies FastAPI and Streamlit behind one ASGI gateway: use FluxLit for routes and pages, ApiClient from Streamlit for server-side HTTP to your API, and FluxLitTestClient in tests (with apptest_select_page() and assert_no_streamlit_exception helpers).

The fluxlit console script (see fluxlit.cli) runs the combined dev/prod stack.

Stable surface: symbols in __all__ below are the primary supported imports for applications and tests; treat other submodules as internal unless documented in Support matrix or API reference.

Optional auth ergonomics (after pip install "fluxlit[auth]"): fluxlit.app.FluxLit.make_jwt_bearer(), fluxlit.app.FluxLit.attach_oidc_login(), and fluxlit.auth.prepare_streamlit_api_client() reduce boilerplate when using FLUXLIT_JWT_* and OIDC BFF env vars.

class fluxlit.Cookie(name)[source]¶

Bases: object

Inject a cookie value by name.

Values come from overrides, set_page_cookie_context(), or st.context.cookies.

class fluxlit.Depends(dependency=None, *, use_cache=True)[source]¶

Bases: object

Declare a page dependency callable ( -> T) resolved before the handler runs.

class fluxlit.FluxLitPublicUrls(fluxlit)[source]¶

Bases: object

Build browser-visible URLs for the Streamlit shell vs the mounted FastAPI API.

App base — origin plus the public mount (FLUXLIT_ROOT_PATH / streamlit_public_path). This is where Streamlit pages live (outside the API prefix).

API base — app base plus api_mount_path (default /api). Liveness, readiness, OpenAPI, and Swagger live under this prefix on the public gateway.

When FLUXLIT_PUBLIC_BASE_URL is set to an absolute URL, it is preferred for the origin (and, if it includes a path, as the app base when that path matches the configured public mount). Otherwise the current request’s host/scheme is used (including X-Forwarded-* when Uvicorn proxy headers are enabled).

api_base(request)[source]¶

Return the browser-visible base URL for the FastAPI app (includes api_mount_path).

Return type:: str

app_base(request)[source]¶

Return the browser-visible base URL for Streamlit (no api_mount_path).

Return type:: str

docs_url(request)[source]¶

Return the Swagger UI URL, or None when docs_url is disabled on api.

Return type:: str | None

for_page(request, path, *, query=None)[source]¶

Return a browser URL for a Streamlit page path (under app_base(), not /api).

path is a URL path such as "/" or "/reports". query values are percent-encoded; use this for deep links to Streamlit pages.

Return type:: str

health_url(request)[source]¶

Return GET liveness URL (.../api/healthz by default).

Return type:: str

openapi_url(request)[source]¶

Return the OpenAPI JSON URL, or None when openapi_url is disabled.

Return type:: str | None

page_url(request, path, *, query=None)[source]¶

Alias for for_page() (invite links, password reset, and other deep links).

Prefer this name in app code and docs when the intent is a shareable page URL rather than an internal path join.

Return type:: str

ready_url(request)[source]¶

Return GET readiness URL (.../api/readyz by default).

Return type:: str

redoc_url(request)[source]¶

Return the ReDoc URL, or None when redoc_url is disabled on api.

Return type:: str | None

class fluxlit.Header(name)[source]¶

Bases: object

Inject a header value by name (from set_page_header_context() or overrides).

class fluxlit.InMemorySessionStore(*, max_entries=5000, default_ttl_seconds=86400.0)[source]¶

Bases: object

Process-local SessionStore (dev / single replica).

Not safe across multiple Uvicorn workers or horizontal replicas without a shared backend.

delete(session_id)[source]¶

Return type:: None

get(session_id)[source]¶

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

set(session_id, data, *, ttl_seconds=None)[source]¶

Persist data for session_id; optional time-to-live in seconds.

TTL semantics: ttl_seconds is None → use default_ttl_seconds (which may itself be None for no expiry). A positive value sets a monotonic deadline. ttl_seconds of 0 does not mean “expire immediately”; it is treated like “no TTL window” for this write (same as a negative value would be if passed, though callers should use None for no expiry).

Return type:: None

class fluxlit.NavigationModel(order=())[source]¶

Bases: object

Order registered pages by URL path (values not registered are ignored).

Paths are compared after strip(); "/" slugifies to "home" in the Streamlit entrypoint the same way as fluxlit.deep_links.match_nav_page().

order: tuple[str, ...] = ()¶

fluxlit.Page¶: alias of PageMeta

class fluxlit.PageMeta(**data)[source]¶

Bases: BaseModel

Optional metadata returned from a page handler or passed to fluxlit.app.FluxLit.page().

Fields mirror streamlit.set_page_config() where applicable. Values returned after the handler runs cannot always update the browser title or layout (Streamlit only allows set_page_config as the first command); use fluxlit.app.FluxLit.page() page_meta= for static per-page config applied at the start of each run. Returned instances still drive sidebar breadcrumbs and the page manifest.

breadcrumb: str | None¶

children: list[dict[str, Any]]¶

description: str | None¶

initial_sidebar_state: Literal['auto', 'expanded', 'collapsed'] | None¶

layout: Literal['centered', 'wide'] | None¶

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

order: int | None¶

page_icon: str | None¶

page_title: str | None¶

class fluxlit.Query(*, description='')[source]¶

Bases: object

Optional metadata for a query-shaped parameter (manifest / docs).

class fluxlit.SessionModel(st, *, extra='ignore')[source]¶

Bases: object

Read/write a Pydantic model to Streamlit session state keys matching field names.

read_into(model_type)[source]¶

Build model_type from st.session_state (missing keys use defaults).

Return type:: TypeVar(ModelT, bound= BaseModel)

write_from(model)[source]¶

Write model.model_dump() into st.session_state.

Return type:: None

class fluxlit.SessionStore(*args, **kwargs)[source]¶

Bases: Protocol

Server-side persistence for URL-bound session blobs (JSON-serializable dicts).

delete(session_id)[source]¶

Remove session_id from the store if present.

Return type:: None

get(session_id)[source]¶

Return the stored dict or None if missing/expired.

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

set(session_id, data, *, ttl_seconds=None)[source]¶

Persist data for session_id; optional time-to-live in seconds.

Return type:: None

fluxlit.apptest_assert_no_errors(at)[source]¶

Raise AssertionError if AppTest captured st.exception or st.error.

Use after FluxLitTestClient.streamlit() (or any AppTest run) to fail fast when the script surfaced an error to users.

Return type:: None

fluxlit.apptest_select_page(at, client, *, target, page, internal_api_base=None, extra_sys_path=None, page_key='page', page_overrides=None)[source]¶

Set page query key and run() with the same FLUXLIT_* patch as streamlit.

AppTest.run does not automatically keep FLUXLIT_* variables from the first FluxLitTestClient.streamlit() call; this helper re-applies the same patch FluxLitTestClient.streamlit() uses so the entrypoint can resolve target again after you change at.query_params.

Return type:: Any

fluxlit.assert_no_streamlit_exception(at)[source]¶

Alias for apptest_assert_no_errors() (Streamlit AppTest naming).

Return type:: None

fluxlit.ensure_url_session(st, store, *, param=None, initial=None, ttl_seconds=None)[source]¶

Ensure st.query_params[param] exists; mint id, seed store, set query param.

Returns the session id (existing or new). If the query param cannot be set (read-only query_params), still returns a new id and persists initial so callers can surface a warning or use client-side navigation.

Return type:: str

fluxlit.hydrate_url_session(st, store, *, param=None, merge=True, blob=None)[source]¶

If st.query_params[param] is set, load the store payload into st.session_state.

merge (default): session_state.setdefault(k, v) for each key in the blob so in-flight widget state wins over stale store keys on first paint.
If the param is missing, returns None and does nothing.
If the param is present but the store has no entry, returns the session id without mutating session_state (caller may seed with SessionStore.set()).

Returns the session id string when the query param is present, else None.

Return type:: str | None

fluxlit.hydrate_url_session_typed(st, store, model, *, param=None, merge=True, strict=False)[source]¶

Like hydrate_url_session(), then validate the store payload as model.

Validates the store blob before merging into st.session_state so invalid payloads do not partially hydrate the UI session.

Returns (session_id_or_none, validated_model_or_none). On validation failure, when strict is false, returns (sid, None); when strict is true, raises pydantic.ValidationError.

Return type:: tuple[str | None, Optional[TypeVar(ModelT, bound= BaseModel)]]

fluxlit.new_session_id()[source]¶

Return a new URL-safe opaque session identifier (≥128 bits).

Return type:: str

fluxlit.parse_query_params(st, model, *, strict=False)[source]¶

Build model from st.query_params.

For multi-value keys, Streamlit may expose a list: a single-element list is coerced to that element; multiple values are passed through as a list.

On pydantic.ValidationError, calls st.error with a short message unless strict is true, in which case the exception is re-raised.

Return type:: TypeVar(ModelT, bound= BaseModel)

fluxlit.persist_url_session(st, store, *, param=None, ttl_seconds=None)[source]¶

Write current st.session_state (shallow dict copy) to the store for param id.

Returns the session id if the param is present, else None.

Values are copied best-effort; remote stores should JSON-encode—non-JSON-safe values may need app-side filtering before SessionStore.set().

If snapshotting session_state fails, the store is not updated but the session id is still returned; a warning is logged (with traceback when FLUXLIT_DEBUG=1).

Return type:: str | None

fluxlit.reset_trace_hook(token)[source]¶

Reset the tracing hook installed by set_trace_hook().

Return type:: None

fluxlit.set_trace_hook(hook)[source]¶

Install a tracing hook for the current context and return a reset token.

Return type:: Token[Callable[[str, Mapping[str, str | int | float | bool | None]], AbstractContextManager[None]] | None]

fluxlit.streamlit_main_path()[source]¶

Return FluxLit’s supported Streamlit entry script for AppTest.from_file.

Use this instead of constructing a path from fluxlit.__file__; the helper is part of FluxLit’s testing API and can keep working if the internal package layout changes.

Return type:: Path

fluxlit.trace_span(name, attributes)[source]¶

Return a context manager for an optional span.

If no hook is installed, this is a no-op. Hook exceptions intentionally bubble up so integrations fail loudly during development instead of hiding broken instrumentation.

Return type:: AbstractContextManager[None]

The FluxLit application object: FastAPI (.api) plus Streamlit pages.

class fluxlit.app.FluxLit(*, title=None, settings=None, import_target=None, fastapi_kwargs=None, streamlit_run_args=None, streamlit_page_config=None, session_store=None)[source]¶

Bases: Generic[SettingsT]

Combine a FastAPI application and registered Streamlit pages in one object.

Use api for HTTP routes, dependencies, and OpenAPI (mounted under api_mount_path on the public gateway). Use urls for browser-visible links (app root vs API prefix, health, docs). Use page() or discover_pages() to register Streamlit UI; the runtime builds st.navigation from registered pages.

Security (optional): make_jwt_bearer() reads FLUXLIT_JWT_* from settings; attach_oidc_login() registers OIDC BFF routes (call at most once per instance) using FLUXLIT_PUBLIC_BASE_URL and FLUXLIT_OIDC_BFF_SECRET when you do not pass secrets explicitly.

GET /healthz (liveness) and GET /readyz (readiness vs Streamlit) are registered on api (hidden from OpenAPI).

Parameters:

title (str | None) – If set, overrides FluxlitSettings.title for this instance.
settings (FluxlitSettings | None) – Explicit settings; default is FluxlitSettings loaded from env / .env.
import_target (str | None) – Optional module:attr string for this instance (same as FLUXLIT_APP / fluxlit.toml target). Use when the module name is not app (e.g. main:app) so uvicorn and the Streamlit child import the right object. If unset, resolution follows env and project file; see fluxlit.runtime.resolve_import_target_for_unified().
fastapi_kwargs (dict[str, Any] | None) – Extra keyword arguments forwarded to fastapi.FastAPI (lifespan, dependencies, openapi_url, docs_url, etc.). title and root_path always come from FluxlitSettings so the API stays aligned with the gateway and Streamlit public path.
streamlit_run_args (Sequence[str] | None) – Optional extra streamlit run CLI tokens; appended to streamlit_run_cli_args on settings.
streamlit_page_config (dict[str, TypeAliasType] | None) – Optional keys merged into streamlit_page_config for streamlit.set_page_config() in the Streamlit process.
session_store (SessionStore | None) – Optional SessionStore for SessionStore-annotated page parameters.

attach_oidc_login(oidc, *, first_party_secret=None, **bff_overrides)[source]¶

Uses public_base_url for redirects unless you pass public_base_url=... in bff_overrides. The first-party JWT signing secret comes from first_party_secret or oidc_bff_secret (FLUXLIT_OIDC_BFF_SECRET).

Returns the fastapi.APIRouter that was included (same as fluxlit.oidc.register_oidc_bff_routes()).

Raises:: ValueError – If this method is called more than once on the same FluxLit instance (duplicate auth routes).
Return type:: APIRouter

build_page_manifest(*, version=1)[source]¶

JSON page manifest; see fluxlit.pages.manifest.build_page_manifest().

Return type:: dict[str, Any]

discover_pages(directory, *, package)[source]¶

Load Streamlit page modules and call register(self) on each.

Imports the subpackage {package}.{directory}, then every submodule (skipping packages and names starting with _). If a module defines register(app: FluxLit) -> None, it is invoked; implementations typically attach handlers with page() inside register.

Registered pages are sorted by (path, title) for stable navigation order.

Parameters:

directory (str) – Subpackage name under package (e.g. "pages").
package (str) – Importable parent package (must have __path__).

Return type:

Self

Returns:

self for chaining.

Raises:

TypeError – If package is not a package.
ImportError – If {package}.{directory} cannot be imported.

Note

If a page module’s register(app) raises after earlier modules ran, pages from those modules remain registered (best-effort; no rollback).

get_client()[source]¶

Return an ApiClient for server-side API calls.

Uses FLUXLIT_INTERNAL_API_BASE when set (as in the managed runtime). When debug is true, the client forwards X-Request-ID to the API for log correlation with the gateway.

Return type:: ApiClient

make_jwt_bearer()[source]¶

JWT JWTBearer from settings (FLUXLIT_JWT_*).

Requires jwt_issuer, jwt_audience, and either jwt_hs256_secret or jwt_jwks_url. Raises ValueError with env-oriented hints if misconfigured.

Return type:: JWTBearer

navigation(model)[source]¶

Set optional sidebar ordering for registered pages (by URL path).

Return type:: Self

page(path, *, title=None, icon=None, tags=None, page_meta=None)[source]¶

Decorator registering a Streamlit page at a URL path.

The decorated callable should accept (st, client) where st is the Streamlit module and client is an ApiClient for your mounted API. Additional parameters may be injected (see docs/streamlit-pages-typing.md).

Parameters:

path (str) – URL path segment for Streamlit (e.g. "/", "/reports").
title (str | None) – Sidebar / navigation title; defaults from the function name.
icon (str | None) – Optional icon for streamlit.navigation / st.Page.
tags (Sequence[str] | None) – Optional tags for manifests and tooling.
page_meta (PageMeta | None) – Static PageMeta merged into st.Page where Streamlit supports it (e.g. page_icon as icon).

Return type:

Callable[[Callable[..., Any]], Callable[..., Any]]

Returns:

Decorator that registers the function and returns it unchanged.

property page_records: list[PageRecord]¶: Full PageRecord entries including tags and metadata.

property pages: list[tuple[str, str, Callable[[...], Any]]]¶: (path, title, handler) tuples (backward compatible).

property urls: FluxLitPublicUrls¶: Browser-visible URL helpers for this app (app root vs api_mount_path).

HTTP client for calling the FastAPI app from Streamlit (server-side, same process host).

Typing mirrors httpx.Client via httpx._types / httpx._client (see tests/test_httpx_import_contract.py); re-run tests after httpx upgrades.

class fluxlit.client.ApiClient(base_url=None, *, timeout=30.0, default_headers=None, auth_header_factory=None, propagate_request_id=False)[source]¶

Bases: object

Sync HTTPX client scoped to your gateway-mounted API.

Paths are relative to the API base (including the /api prefix in the base URL). For example use client.get("/users"), not client.get("/api/users"). Paths must not be absolute URLs (http://...) or scheme-relative (//...); those are rejected with ValueError so user-controlled strings cannot bypass base_url to another host.

Use for_fluxlit() or with_bearer() when routes require Authorization.

The runtime sets FLUXLIT_INTERNAL_API_BASE (e.g. http://127.0.0.1:8000/api) for Streamlit subprocesses so defaults work without passing base_url.

close()[source]¶

Close the underlying HTTPX client.

Return type:: None

delete(path, *, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

DELETE request.

Return type:: httpx.Response

classmethod for_fluxlit(*, bearer_token=None, auth_header_factory=None, base_url=None, timeout=30.0, default_headers=None, propagate_request_id=False)[source]¶

Convenience constructor with static bearer token or factory (mutually exclusive).

Return type:: ApiClient

get(path, *, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

GET request.

Return type:: httpx.Response

get_model(path, model, *, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

GET and parse JSON into a Pydantic model (raises on 4xx/5xx or validation).

Parameters:

path (str) – Relative API path.
model (type[T]) – Pydantic model type for the response body.
kwargs (Remaining) – forwarded to get().

Return type:

post(path, *, content=None, data=None, files=None, json=None, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

POST request.

Return type:: httpx.Response

post_model(path, response_model, *, body=None, content=None, data=None, files=None, json=None, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

POST JSON body and parse the response as response_model.

Parameters:

path (str) – Relative API path.
response_model (type[T]) – Pydantic model type for the response body.
body (BaseModel | Mapping[str, JsonValue] | None) – Request JSON (from model_dump() if a BaseModel). Mutually exclusive with json.
kwargs (Remaining) – forwarded to post().

Return type:

put(path, *, content=None, data=None, files=None, json=None, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

PUT request.

Return type:: httpx.Response

request(method, path, *, content=None, data=None, files=None, json=None, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

Send a request; path may omit a leading slash.

Raises:: ValueError – If path is an absolute or scheme-relative URL (see class doc).
Return type:: httpx.Response

with_bearer(bearer_token)[source]¶

Return a new client with the same base URL and options plus Authorization: Bearer.

Use on the injected page client (no auth by default) when you already have a token in st.session_state and want the same ApiClient surface without calling for_fluxlit(). If this client already has an auth_header_factory, the new factory merges those headers first, then sets Authorization to this bearer (call-time headers still win on key clash).

The returned client owns a separate httpx connection pool; close it when done or use a with block.

Return type:: ApiClient

Configuration: settings, project file defaults, and JSON value typing.

class fluxlit.config.FluxlitSettings(**data)[source]¶

Bases: BaseSettings

Runtime configuration for FluxLit, CLI defaults, and FastAPI construction.

Values are read from environment variables prefixed with FLUXLIT_ and from a .env file in the working directory if present. Unknown env keys are ignored.

CLI commands such as fluxlit dev merge these with fluxlit.toml / pyproject [tool.fluxlit] and explicit flags; see fluxlit.config.project for precedence.

Key fields:

title — FastAPI / UX title.
gateway_host / gateway_port — default bind address for Uvicorn.
api_mount_path — public URL prefix for the API (default /api).
root_path — ASGI root when behind a reverse proxy.
enable_request_logging — per-request INFO logs on the FastAPI app.
enable_gateway_access_log — per-request INFO logs on the gateway (structured extras).
trust_proxy / forwarded_allow_ips — Uvicorn proxy trust (e.g. Posit Connect).
streamlit_public_path — optional subpath when root_path is unset.
streamlit_run_cli_args — extra streamlit run CLI tokens (JSON list in env).
streamlit_page_config — keys forwarded to st.set_page_config (JSON object in env).
cors_middleware_kwargs — extra kwargs for CORSMiddleware when CORS is enabled.
experimental_yield_pages / async_page_depends — experimental; semantics and promotion criteria are in docs/support-matrix (Experimental FluxlitSettings).

api_mount_path: str¶

async_page_depends: bool¶

cors_allow_credentials: bool¶

cors_allow_origins: list[str]¶

cors_middleware_kwargs: dict[str, JsonValue]¶

debug: bool¶

enable_gateway_access_log: bool¶

enable_gateway_prometheus_metrics: bool¶

enable_request_logging: bool¶

enable_security_headers: bool¶

experimental_yield_pages: bool¶

forwarded_allow_ips: str | None¶

gateway_forward_client_headers_to_streamlit: list[str]¶

gateway_host: str¶

gateway_httpx_max_connections: int¶

gateway_httpx_max_keepalive_connections: int¶

gateway_max_concurrent_upstream_http: int¶

gateway_max_proxy_request_body_bytes: int¶

gateway_port: int¶

gateway_prometheus_metrics_path: str¶

gateway_upstream_connect_timeout_s: float¶

gateway_upstream_read_timeout_s: float¶

gateway_ws_close_timeout_s: float | None¶

gateway_ws_max_message_bytes: int | None¶

gateway_ws_open_timeout_s: float¶

gateway_ws_ping_interval_s: float | None¶

gateway_ws_ping_timeout_s: float | None¶

jwt_audience: str¶

jwt_hs256_secret: str¶

jwt_issuer: str¶

jwt_jwks_url: str¶

jwt_leeway_seconds: int¶

log_level: str¶

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'FLUXLIT_', 'env_prefix_target': 'variable', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)¶

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self (BaseModel) – The BaseModel instance.
context (Any) – The context.

Return type:

None

oidc_bff_secret: str¶

public_base_url: str¶

public_mount_path()[source]¶

Browser-visible path prefix (root_path, else streamlit_public_path).

Return type:: str

root_path: str¶

streamlit_host: str¶

streamlit_page_config: dict[str, JsonValue]¶

streamlit_port: int¶

streamlit_public_path: str¶

streamlit_run_cli_args: list[str]¶

strict_page_signatures: bool¶

strict_public_base_url: bool¶

strict_startup: bool¶

title: str¶

trust_proxy: bool¶

url_session_query_param: str¶

uvicorn_graceful_shutdown_timeout_s: float | None¶

type fluxlit.config.JsonValue = _AllowAnyJson]¶

class fluxlit.config.ProjectConfig(target=None, gateway_host=None, gateway_port=None, log_level=None, api_mount_path=None, root_path=None)[source]¶

Bases: object

Structured defaults read from a project file (not environment).

Populated from top-level keys in fluxlit.toml or from [tool.fluxlit] in pyproject.toml. Only known keys are stored; others are ignored.

Fields mirror CLI-related settings: target (e.g. app:app), bind defaults, log_level, and optional api_mount_path / root_path.

api_mount_path: str | None = None¶

gateway_host: str | None = None¶

gateway_port: int | None = None¶

log_level: str | None = None¶

root_path: str | None = None¶

target: str | None = None¶

fluxlit.config.load_project_config(cwd=None)[source]¶

Parse fluxlit.toml or [tool.fluxlit] in pyproject.toml.

If both files exist, fluxlit.toml takes precedence. Returns None if neither file exists, the relevant section is missing, the file is not valid TOML, or the top-level value is not a table (dict).

Parameters:: cwd (Path | None) – Directory to search; defaults to pathlib.Path.cwd().
Return type:: ProjectConfig | None
Returns:: Parsed config, or None when no project file applies.

fluxlit.config.resolve_binding(*, cli_host, cli_port, cli_log_level, pc, settings_gateway_host, settings_gateway_port, settings_log_level)[source]¶

Resolve host, port, and log level: CLI > project file > FluxlitSettings.

settings_* arguments should come from the loaded FluxLit instance (already merged with environment).

Return type:: tuple[str, int, str]

fluxlit.config.resolve_target(cli_target, pc)[source]¶

Pick the app import target: CLI argument, then project file, then app:app.

Return type:: str

Load optional project defaults from fluxlit.toml or pyproject.toml.

class fluxlit.config.project.ProjectConfig(target=None, gateway_host=None, gateway_port=None, log_level=None, api_mount_path=None, root_path=None)[source]¶

Bases: object

Structured defaults read from a project file (not environment).

Populated from top-level keys in fluxlit.toml or from [tool.fluxlit] in pyproject.toml. Only known keys are stored; others are ignored.

Fields mirror CLI-related settings: target (e.g. app:app), bind defaults, log_level, and optional api_mount_path / root_path.

api_mount_path: str | None = None¶

gateway_host: str | None = None¶

gateway_port: int | None = None¶

log_level: str | None = None¶

root_path: str | None = None¶

target: str | None = None¶

fluxlit.config.project.load_project_config(cwd=None)[source]¶

Parse fluxlit.toml or [tool.fluxlit] in pyproject.toml.

If both files exist, fluxlit.toml takes precedence. Returns None if neither file exists, the relevant section is missing, the file is not valid TOML, or the top-level value is not a table (dict).

Parameters:: cwd (Path | None) – Directory to search; defaults to pathlib.Path.cwd().
Return type:: ProjectConfig | None
Returns:: Parsed config, or None when no project file applies.

fluxlit.config.project.resolve_binding(*, cli_host, cli_port, cli_log_level, pc, settings_gateway_host, settings_gateway_port, settings_log_level)[source]¶

Resolve host, port, and log level: CLI > project file > FluxlitSettings.

settings_* arguments should come from the loaded FluxLit instance (already merged with environment).

Return type:: tuple[str, int, str]

fluxlit.config.project.resolve_target(cli_target, pc)[source]¶

Pick the app import target: CLI argument, then project file, then app:app.

Return type:: str

Logging helpers: request IDs, JSON formatters, redaction for safe access logs.

class fluxlit.logging.JsonLogFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)[source]¶

Bases: Formatter

Emit one JSON object per log line, including merged extra fields.

format(record)[source]¶

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

Return type:: str

fluxlit.logging.get_request_id()[source]¶

Return the active request ID from request_id_ctx, if any.

Return type:: str | None

fluxlit.logging.new_request_id()[source]¶

Generate a new opaque request ID (UUID4 string).

Return type:: str

fluxlit.logging.redact_authorization(value)[source]¶

Return a placeholder for an Authorization header value.

Return type:: str

fluxlit.logging.redact_query_string(query_string, *, sensitive_keys=None)[source]¶

Return query_string with listed keys’ values replaced by <redacted>.

query_string is the raw key=value&... portion (no leading ?). Unknown or malformed segments are returned unchanged when parsing fails.

Return type:: str

fluxlit.logging.reset_request_id(token)[source]¶

Restore the previous context after set_request_id() (typically in finally).

Return type:: None

fluxlit.logging.sanitize_headers(headers)[source]¶

Copy headers with Authorization and Cookie values redacted.

Return type:: dict[str, str]

fluxlit.logging.set_request_id(value)[source]¶

Bind value into request_id_ctx; returns a token for reset_request_id().

Return type:: Token[str | None]

ASGI gateway: dispatch api_prefix to FastAPI, proxy everything else to Streamlit.

HTTP requests and WebSockets are forwarded to an upstream base URL (the Streamlit server). The client’s Host header is preserved on the upstream request so Streamlit sees the public gateway host/port (e.g. 127.0.0.1:8777), matching the browser’s Origin; otherwise WebSocket same-origin checks fail and the UI stays blank/black.

Request IDs are taken from X-Request-ID or generated, stored in fluxlit.logging for the duration of each request, and re-sent to Streamlit on proxied HTTP and WebSocket hops as authoritative X-Request-ID (the gateway wins over any client-supplied value on that upstream leg).

Top-level /docs, /redoc, and /openapi.json redirect to the same paths under api_prefix so Swagger is not accidentally proxied to Streamlit (which would look like a blank page).

class fluxlit.gateway.GatewayProxyOptions(connect_timeout=30.0, read_timeout=120.0, max_proxy_body_bytes=0, max_concurrent_upstream_http=0, httpx_max_connections=0, httpx_max_keepalive_connections=0, ws_open_timeout_s=30.0, ws_ping_interval_s=None, ws_ping_timeout_s=None, ws_close_timeout_s=None, ws_max_message_bytes=None, forward_client_headers_http=frozenset({}))[source]¶

Bases: object

Upstream HTTP/WebSocket tuning for fluxlit.gateway.build_gateway().

Mirrors FluxlitSettings gateway fields.

connect_timeout: float = 30.0¶

forward_client_headers_http: frozenset[str] = frozenset({})¶

httpx_max_connections: int = 0¶

httpx_max_keepalive_connections: int = 0¶

max_concurrent_upstream_http: int = 0¶

max_proxy_body_bytes: int = 0¶

read_timeout: float = 120.0¶

ws_close_timeout_s: float | None = None¶

ws_max_message_bytes: int | None = None¶

ws_open_timeout_s: float = 30.0¶

ws_ping_interval_s: float | None = None¶

ws_ping_timeout_s: float | None = None¶

fluxlit.gateway.build_gateway(api_app, upstream_base, *, upstream_resolver=None, access_log=False, api_prefix='/api', root_mount='', proxy_settings=None)[source]¶

Build the composite ASGI application used as Uvicorn’s entrypoint.

Routes whose path equals api_prefix or starts with api_prefix/ are forwarded to api_app with that prefix stripped from path and raw_path. All other HTTP traffic and WebSockets are reverse-proxied to upstream_base (typically the internal Streamlit origin), or to the URL returned by upstream_resolver when that is provided (evaluated per request).

Lifespan events are delegated only to api_app.

Parameters:

api_app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Inner FastAPI / Starlette app (mount path not included in its routes).
upstream_base (str) – Base URL for Streamlit when upstream_resolver is unset.
upstream_resolver (Callable[[], str] | None) – If set, called for each proxied request to get the current upstream base (e.g. after Streamlit restarts on a new port). upstream_base is ignored for proxying when this is set.
access_log (bool) – If True, emit one INFO log per request with structured extra fields.
api_prefix (str) – Public URL prefix for the API (default /api).
root_mount (str) – Optional browser-visible path prefix when the app is published under a subpath (e.g. Posit Connect / Workbench). Must match root_path and Streamlit server.baseUrlPath. When the proxy forwards the full path, this strip is applied before dispatch; Streamlit still receives paths that include the prefix.
proxy_settings (FluxlitSettings | None) – Optional FluxlitSettings for upstream timeouts, body limits, concurrency, and WebSocket tuning (defaults match historical hardcoded gateway behavior when omitted).

Return type:

Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

Returns:

A callable ASGI3 application.

fluxlit.gateway.normalize_api_mount_path(api_mount_path)[source]¶

Normalize the public API URL prefix for gateway dispatch and URL helpers.

Ensures a single leading / and no trailing slash (except the root "/"). Empty or whitespace-only values fall back to "/api".

Return type:: str

fluxlit.gateway.normalize_root_mount(raw)[source]¶

Normalize a public URL prefix (e.g. Posit Connect content path) for routing.

Returns "" when unset, otherwise a path starting with / and no trailing slash (except root "/" is not used — empty means no mount).

Return type:: str

fluxlit.gateway.split_gateway_paths(path, root_mount)[source]¶

Split the ASGI path for dispatch vs Streamlit upstream.

Some reverse proxies forward the full public path (/content/123/api/...). Others strip the mount and only forward the suffix (/api/...) while setting ASGI root_path. normalize_root_mount() should match the browser-visible prefix configured for Streamlit server.baseUrlPath.

Return type:: tuple[str, str]
Returns:: (dispatch_path, streamlit_path) — use dispatch_path to choose API vs Streamlit; send streamlit_path to the Streamlit sidecar when proxying.

Readiness probes for the unified runtime (Streamlit sidecar).

async fluxlit.health.probe_streamlit_ready(*, timeout_s=0.5, upstream=None, settings=None)[source]¶

Return (ok, detail) for whether the Streamlit upstream accepts HTTP traffic.

When FLUXLIT_STREAMLIT_UPSTREAM / file state is unset (typical in bare FastAPI tests), returns (True, "not_configured").

When configured, readiness requires an HTTP 2xx response from GET on at least one candidate URL (see _streamlit_readiness_urls()). Other status codes are treated as not ready. When multiple URLs are probed and all fail, detail aggregates path and status (for example upstream_http_failed /:404;/myapp/:503).

Return type:: tuple[bool, str]

Helpers to redact secrets from header dicts before logging or tracing.

fluxlit.logging.redact.redact_authorization(value)[source]¶

Return a placeholder for an Authorization header value.

Return type:: str

fluxlit.logging.redact.redact_query_string(query_string, *, sensitive_keys=None)[source]¶

Return query_string with listed keys’ values replaced by <redacted>.

query_string is the raw key=value&... portion (no leading ?). Unknown or malformed segments are returned unchanged when parsing fails.

Return type:: str

fluxlit.logging.redact.sanitize_headers(headers)[source]¶

Copy headers with Authorization and Cookie values redacted.

Return type:: dict[str, str]

Process orchestration: load FluxLit by import path, spawn Streamlit, run Uvicorn.

fluxlit.runtime.asgi_from_fluxlit(fl, import_target)[source]¶

Build unified ASGI (gateway + Streamlit) for an existing FluxLit.

Use when you already hold the instance (e.g. uvicorn main:app). import_target must be the module:attr the Streamlit subprocess imports; use fluxlit.runtime.resolve_import_target_for_unified() or set import_target / fluxlit.toml / FLUXLIT_APP.

Behavior matches create_unified_app() (lifespan, sidecar, inner FastAPI hooks).

Return type:: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

fluxlit.runtime.create_gateway_app()[source]¶

ASGI factory for Uvicorn --factory reload mode.

Reads FLUXLIT_APP (import target), Streamlit upstream from FLUXLIT_STREAMLIT_UPSTREAM_FILE or FLUXLIT_STREAMLIT_UPSTREAM, and FLUXLIT_API_PREFIX from the environment, then returns fluxlit.gateway.build_gateway() over the loaded FastAPI app.

Return type:: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]
Returns:: An ASGI3 callable (same contract as build_gateway()).
Raises:: RuntimeError – If FLUXLIT_APP is unset or the upstream URL cannot be resolved.

fluxlit.runtime.create_unified_app()[source]¶

ASGI factory for Uvicorn --factory (same stack as fluxlit.app.FluxLit.__call__()).

Requires FLUXLIT_APP=module:attr. Prefer uvicorn main:app when app is a FluxLit instance so you do not need this factory or env indirection.

Return type:: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

fluxlit.runtime.default_pidfile_path(explicit=None)[source]¶

Path for fluxlit dev|run PID file (current directory unless overridden).

Return type:: Path

fluxlit.runtime.find_free_port()[source]¶

Bind to 127.0.0.1:0 and return the assigned ephemeral port.

Return type:: int

fluxlit.runtime.internal_api_base_url(*, bind_host, port, api_mount_path)[source]¶

Build FLUXLIT_INTERNAL_API_BASE for the Streamlit child (same machine as gateway).

bind_host is the Uvicorn bind address; the URL uses a loopback-safe host when needed so ApiClient can connect from the sidecar process.

Return type:: str

fluxlit.runtime.load_fluxlit(target)[source]¶

Import module:attribute and ensure the object is a FluxLit.

Raises:

ValueError – If target is not a module:attr string.
ModuleNotFoundError – If module cannot be imported (message includes a short hint).
AttributeError – If module has no such attribute.
TypeError – If attr is not a FluxLit instance.

Return type:

FluxLit[Any]

fluxlit.runtime.read_streamlit_upstream_url()[source]¶

Current Streamlit base URL from write_streamlit_upstream_state() or plain env.

Return type:: str

fluxlit.runtime.resolve_import_target_for_unified(fl)[source]¶

Resolve module:attr for the Streamlit child and import stability.

Order: import_target, FLUXLIT_APP, then project file (fluxlit.toml / pyproject.toml), then app:app.

Return type:: str

fluxlit.runtime.run_unified(target, *, host='127.0.0.1', port=8000, reload=False, reload_scope='gateway', log_level='info', proxy_headers=False, forwarded_allow_ips=None, pidfile=None, write_pidfile=True, workbench_mode=False)[source]¶

Start Streamlit on a free localhost port and Uvicorn on host:port.

Sets process environment so create_gateway_app / Streamlit entry can resolve the app and internal API base (loopback-safe URL derived from host, port, and api_mount_path). If Streamlit exits, the gateway is stopped. On shutdown, the Streamlit child receives SIGINT / terminate / kill (platform-dependent).

Parameters:

target (str) – module:fluxlit_instance import path.
host (str) – Uvicorn bind host.
port (int) – Uvicorn bind port (public).
reload (bool) – If True, use Uvicorn reload with create_gateway_app().
reload_scope (str) – gateway (default) or full. full also restarts Streamlit when watched files change (requires watchfiles); see CLI --reload-scope.
log_level (str) – Uvicorn log level.
proxy_headers (bool) – Forwarded to uvicorn.Config.
forwarded_allow_ips (str | None) – Forwarded to uvicorn.Config.
pidfile (Path | None) – Optional explicit path for the PID file (see default_pidfile_path()).
write_pidfile (bool) – If False, do not create a PID file (also skipped when FLUXLIT_NO_PIDFILE is 1 / true / yes).
workbench_mode (bool) – If True, print a Workbench/Connect-oriented startup banner and force proxy_headers on for Uvicorn (still merge forwarded_allow_ips).

Return type:

None

fluxlit.runtime.shutdown_unified_process(pidfile=None, *, force=False, wait_s=5.0)[source]¶

Stop a stack started by fluxlit.runtime.run_unified() using its PID file.

Sends SIGTERM to the recorded PID (the process running Uvicorn + supervision). On Windows, uses taskkill /T (and /F when force is True) instead of os.kill, which does not reliably terminate arbitrary processes.

If force is True on POSIX, sends SIGKILL after wait_s if still running.

Return type:: tuple[int, str]
Returns:: (exit_code, message) where exit_code is 0 on success, 1 on failure (still running after timeout / permission error), 2 if the pidfile is missing.

fluxlit.runtime.update_streamlit_upstream_file(path, url)[source]¶

Replace the on-disk upstream URL (e.g. after restarting Streamlit on a new port).

Return type:: None

fluxlit.runtime.write_streamlit_upstream_state(url)[source]¶

Write url to a temp file and set env for the gateway resolver and subprocesses.

Return type:: Path

Test helpers: gateway-scoped HTTP client and Streamlit AppTest integration.

class fluxlit.testing.FluxLitTestClient(app, api_prefix='/api', root_mount='')[source]¶

Bases: object

Test harness that mirrors production routing (API prefix + gateway).

API — api is a Starlette TestClient wired through fluxlit.gateway.build_gateway(), so paths include the configured api_prefix and /healthz behaves like production.

Public mount — Set root_mount (or with_root_path()) to simulate Workbench / Posit-style URLs where the browser path is {mount}{api_prefix}/…. Use api_get() / api_post() so URLs are built correctly; optional root_path= on a single call builds a matching gateway for that request only.

Streamlit — streamlit() runs AppTest against fluxlit.streamlit.main with the same environment variables the runtime sets.

AppTest multipage — assert_no_streamlit_exception() fails on st.error / st.exception. select_page() sets ?page= and reruns; the entrypoint honors fluxlit.deep_links.match_nav_page() before streamlit.navigation(). Module-level apptest_assert_no_errors(), assert_no_streamlit_exception(), and apptest_select_page() mirror the same behavior.

property api: TestClient¶: HTTP test client; non-API routes hit a dummy upstream (unused for /api tests).

api_get(path, *, root_path=None, **kwargs)[source]¶

GET relative to api_prefix (leading slash optional on path).

When root_path is set, the request uses that public mount for this call only (a separate gateway instance). Omit it to use root_mount on this client.

Return type:: Response

api_post(path, *, root_path=None, **kwargs)[source]¶

POST relative to api_prefix (optional per-call root_path).

Return type:: Response

api_prefix: str = '/api'¶

app: FluxLit[Any]¶

assert_docs_available(*, root_path=None)[source]¶

Assert OpenAPI JSON and Swagger UI are reachable through the gateway.

Raises:: AssertionError – If OpenAPI is missing/invalid or GET …/docs is not available (including when FastAPI docs_url is disabled).
Return type:: None

assert_no_streamlit_exception(at)[source]¶

Fail if at collected st.exception or st.error blocks.

Return type:: None

openapi()[source]¶

Fetch and parse GET {api_prefix}/openapi.json; raises if not a JSON object.

Return type:: dict[str, Any]

root_mount: str = ''¶

select_page(at, page, *, target, internal_api_base=None, extra_sys_path=None, page_key='page', page_overrides=None)[source]¶

Set query key and run() again with target env (same patch as streamlit).

Return type:: Any

streamlit(*, target, internal_api_base=None, extra_sys_path=None, query_params=None, page_overrides=None)[source]¶

Execute Streamlit’s AppTest against fluxlit.streamlit.main.

Requires Streamlit >= 1.30 for AppTest. Patches FLUXLIT_APP, FLUXLIT_INTERNAL_API_BASE, FLUXLIT_API_PREFIX, and FLUXLIT_TESTS for the duration of the run.

Parameters:

target (str) – Import path module:FluxLit (same as CLI).
internal_api_base (str | None) – Override internal API URL; default is a placeholder with the correct api_prefix suffix.
extra_sys_path (str | Path | None) – Optional directory prepended to sys.path (e.g. project root).
query_params (dict[str, str] | None) – Optional initial query string values (same as assigning to AppTest.query_params before the first run()).
page_overrides (dict[str, Any] | None) – Optional JSON-serializable map merged into handler dependency injection (via FLUXLIT_TEST_PAGE_OVERRIDES) for Header / test doubles.

Return type:

Any

Returns:

The result of AppTest.from_file(...).run() (Streamlit type).

with_root_path(root_path)[source]¶

Return a copy of this client with a browser-visible path prefix (Workbench-style).

Return type:: FluxLitTestClient

fluxlit.testing.apptest_assert_no_errors(at)[source]¶

Raise AssertionError if AppTest captured st.exception or st.error.

Use after FluxLitTestClient.streamlit() (or any AppTest run) to fail fast when the script surfaced an error to users.

Return type:: None

fluxlit.testing.apptest_select_page(at, client, *, target, page, internal_api_base=None, extra_sys_path=None, page_key='page', page_overrides=None)[source]¶

Set page query key and run() with the same FLUXLIT_* patch as streamlit.

Return type:: Any

fluxlit.testing.assert_no_streamlit_exception(at)[source]¶

Alias for apptest_assert_no_errors() (Streamlit AppTest naming).

Return type:: None

fluxlit.testing.streamlit_main_path()[source]¶

Return FluxLit’s supported Streamlit entry script for AppTest.from_file.

Use this instead of constructing a path from fluxlit.__file__; the helper is part of FluxLit’s testing API and can keep working if the internal package layout changes.

Return type:: Path

Optional no-dependency tracing hooks for FluxLit internals.

The core package does not depend on OpenTelemetry. Integrations can register a small context-manager factory and translate FluxLit spans into any tracing backend.

fluxlit.tracing.reset_trace_hook(token)[source]¶

Reset the tracing hook installed by set_trace_hook().

Return type:: None

fluxlit.tracing.set_trace_hook(hook)[source]¶

Install a tracing hook for the current context and return a reset token.

Return type:: Token[Callable[[str, Mapping[str, str | int | float | bool | None]], AbstractContextManager[None]] | None]

fluxlit.tracing.trace_span(name, attributes)[source]¶

Return a context manager for an optional span.

If no hook is installed, this is a no-op. Hook exceptions intentionally bubble up so integrations fail loudly during development instead of hiding broken instrumentation.

Return type:: AbstractContextManager[None]

Small utilities for declaring fastapi.APIRouter instances.

fluxlit.api.router(*, prefix='', tags=None)[source]¶

Create an APIRouter with optional prefix and OpenAPI tags.

Exists to normalize tags typing (list[str] vs enum tags) for strict callers.

Parameters:

prefix (str) – Path prefix for all routes on this router.
tags (Sequence[str] | None) – Tag names attached to operations in OpenAPI.

Return type:

APIRouter

Returns:

A new APIRouter.

Authentication helpers (JWT, OIDC BFF, Streamlit-side helpers, trusted proxy headers).

class fluxlit.auth.GenericOIDCClient(config)[source]¶

Bases: object

OIDC provider using /.well-known/openid-configuration (Authorization Code + PKCE).

authorization_url(*, redirect_uri, state, code_challenge, scope=None)[source]¶

Return type:: str

exchange_code(*, code, code_verifier, redirect_uri)[source]¶

Return type:: dict[str, Any]

property issuer: str¶

load_discovery_sync()[source]¶

Return type:: None

class fluxlit.auth.GenericOIDCClientConfig(issuer, client_id, client_secret, http_timeout=30.0)[source]¶

Bases: object

client_id: str¶

client_secret: str¶

http_timeout: float = 30.0¶

issuer: str¶

class fluxlit.auth.InMemoryOIDCBFFTokenStore(_pkce=<factory>, _exchange=<factory>, _lock=<factory>, state_ttl_seconds=600.0, otc_ttl_seconds=120.0)[source]¶

Bases: object

Process-local PKCE and exchange storage (default for OIDCBFFConfig).

otc_ttl_seconds: float = 120.0¶

pop_exchange_token(auth_code, *, now)[source]¶

Return type:: str | None

pop_pkce_verifier(state, *, now)[source]¶

Return type:: str | None

save_exchange_token(auth_code, access_token, *, now)[source]¶

Return type:: None

save_pkce_verifier(state, code_verifier, *, now)[source]¶

Return type:: None

state_ttl_seconds: float = 600.0¶

class fluxlit.auth.JWTAuthConfig(issuer, audience, algorithms=<factory>, jwks_url=None, hs256_secret=None, leeway_seconds=0)[source]¶

Bases: object

Configuration for JWTBearer.

algorithms: list[str]¶

audience: str | list[str]¶

hs256_secret: str | None = None¶: If set, tokens are validated with HS256 using this secret (development).

issuer: str¶

jwks_url: str | None = None¶: JWKS URI (required for asymmetric algorithms unless using HS256-only).

leeway_seconds: int = 0¶

class fluxlit.auth.JWTBearer(config)[source]¶

Bases: object

FastAPI dependency: validate Authorization: Bearer and return StandardClaims.

classmethod from_fluxlit_settings(settings)[source]¶

Build a bearer dependency from settings / FLUXLIT_JWT_*.

Uses FluxlitSettings. Provide either jwt_hs256_secret (development) or jwt_jwks_url (RS256 / JWKS), plus jwt_issuer and jwt_audience. Error messages name the env vars.

Return type:: JWTBearer

class fluxlit.auth.OIDCBFFConfig(oidc, first_party_secret, token_issuer='fluxlit-bff', token_audience='fluxlit-app', access_token_ttl_seconds=3600, public_base_url='', login_path='/auth/login', callback_path='/auth/callback', exchange_path='/auth/exchange', scope='openid profile email', streamlit_redirect_path='/', state_ttl_seconds=600, otc_ttl_seconds=120, id_token_audience='', id_token_leeway_seconds=0, allow_unverified_id_token_for_custom_oidc=False, bff_token_store=None)[source]¶

Bases: object

Settings for register_oidc_bff_routes().

access_token_ttl_seconds: int = 3600¶

allow_unverified_id_token_for_custom_oidc: bool = False¶

If True, allow non-GenericOIDCClient providers to use parse-only sub.

Default False: custom OIDCProvider callbacks reject minting tokens from id_token without JWKS verification. Set True only when the provider verifies tokens before returning them, or for tests.

bff_token_store: OIDCBFFTokenStore | None = None¶

Optional shared storage for PKCE state and one-time auth_code values.

When None, register_oidc_bff_routes() uses InMemoryOIDCBFFTokenStore with state_ttl_seconds and otc_ttl_seconds (single replica).

callback_path: str = '/auth/callback'¶

exchange_path: str = '/auth/exchange'¶

first_party_secret: str¶

id_token_audience: str = ''¶

Expected aud for id_token validation.

If empty, client_id from GenericOIDCClient is used.

id_token_leeway_seconds: int = 0¶: Clock skew leeway (seconds) when validating id_token with JWKS.

login_path: str = '/auth/login'¶

oidc: OIDCProvider¶

otc_ttl_seconds: int = 120¶

public_base_url: str = ''¶: Origin only, e.g. https://app.example.com (no trailing slash).

scope: str = 'openid profile email'¶

state_ttl_seconds: int = 600¶

streamlit_redirect_path: str = '/'¶: Browser path to send the user to with a one-time auth_code query param.

token_audience: str = 'fluxlit-app'¶

token_issuer: str = 'fluxlit-bff'¶

class fluxlit.auth.OIDCBFFTokenStore(*args, **kwargs)[source]¶

Bases: Protocol

Storage for OIDC BFF PKCE state and one-time auth_code exchange payloads.

Defaults to InMemoryOIDCBFFTokenStore (process memory). Multi-replica API deployments should inject a shared implementation (for example Redis with TTL).

pop_exchange_token(auth_code, *, now)[source]¶

Return type:: str | None

pop_pkce_verifier(state, *, now)[source]¶

Return type:: str | None

save_exchange_token(auth_code, access_token, *, now)[source]¶

Return type:: None

save_pkce_verifier(state, code_verifier, *, now)[source]¶

Return type:: None

class fluxlit.auth.OIDCProvider(*args, **kwargs)[source]¶

Bases: Protocol

Minimal OIDC surface for discovery and token exchange.

authorization_url(*, redirect_uri, state, code_challenge, scope=None)[source]¶

Return type:: str

exchange_code(*, code, code_verifier, redirect_uri)[source]¶

Return type:: dict[str, Any]

property issuer: str¶

class fluxlit.auth.RequireRoles(bearer, *roles, roles_claim='roles')[source]¶

Bases: object

Dependency factory: require at least one role from a claim (string or list).

class fluxlit.auth.RequireScopes(bearer, *scopes, scope_claim='scope')[source]¶

Bases: object

Dependency factory: require OAuth2-style scopes (space-delimited or list claim).

class fluxlit.auth.StandardClaims(**data)[source]¶

Bases: BaseModel

Common JWT claims plus extra keys from the token payload.

aud: str | list[str] | None¶

exp: int | None¶

iat: int | None¶

iss: str | None¶

jti: str | None¶

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

nbf: int | None¶

scope: str | None¶

sub: str | None¶

class fluxlit.auth.TrustedProxyUser(config=None)[source]¶

Bases: object

FastAPI dependency: trusted gateway user header with optional safety checks.

Use only when a network path guarantees that clients cannot spoof the header (for example, the app listens only on loopback and nginx strips inbound identity headers from untrusted clients).

class fluxlit.auth.TrustedProxyUserConfig(header_name='X-Remote-User', require_https=False, trusted_client_hosts=None, require_non_empty_user=True)[source]¶

Bases: object

Policy for TrustedProxyUser.

header_name: str¶

require_https: bool¶: If True, require X-Forwarded-Proto: https or direct https URL scheme.

require_non_empty_user: bool¶: If True, missing header yields 401 instead of allowing empty identity.

trusted_client_hosts: frozenset[str] | None¶: If set, request.client.host must be in this set (use with reverse proxies).

fluxlit.auth.bearer_headers_from_session(st_module, *, session_key='fluxlit_access_token')[source]¶

Build Authorization headers from st.session_state (no logging of secrets).

Return type:: dict[str, str]

fluxlit.auth.exchange_auth_code_from_query(st_module, client, *, exchange_path='/auth/exchange', query_key='auth_code', session_key='fluxlit_access_token')[source]¶

If query_key is present, exchange it for an access token and store in session.

Returns the access token when exchanged; otherwise None. On HTTP errors from the exchange endpoint, raises httpx.HTTPStatusError.

Return type:: str | None

fluxlit.auth.issue_hs256_access_token(*, subject, issuer, audience, secret, ttl_seconds, extra_claims=None)[source]¶

Mint a short-lived HS256 JWT (BFF / dev only — keep secret server-side).

Return type:: str

fluxlit.auth.pkce_pair()[source]¶

Return (code_verifier, code_challenge) for PKCE (S256).

Return type:: tuple[str, str]

fluxlit.auth.prepare_streamlit_api_client(st_module, *, exchange_path='/auth/exchange', session_key='fluxlit_access_token', **client_options)[source]¶

One-step Streamlit helper: exchange auth_code (if present), return an ApiClient.

If a bearer token exists in session_key after the exchange (or was already there), returns ApiClient.for_fluxlit() so protected routes work. Otherwise returns an unauthenticated client.

Return type:: ApiClient

fluxlit.auth.proxy_user_header(header_name='X-Remote-User')[source]¶

Build a dependency that reads a trusted user id from an HTTP header.

Intended for deployments where an upstream proxy (SSO, API gateway) authenticates the user and forwards identity via a header. This does not validate signatures or sessions; it only exposes the header value to route handlers.

For stricter control (HTTPS, client IP allowlists), use TrustedProxyUser.

Parameters:: header_name (str) – Header to read (default X-Remote-User).
Return type:: Callable[[Request], Any]
Returns:: Async dependency returning str | None (the raw header value).

fluxlit.auth.register_oidc_bff_routes(app, config, *, router_prefix='')[source]¶

Attach login, OAuth callback, and Streamlit-friendly token exchange routes.

Return type:: APIRouter

JWT bearer validation for FastAPI (requires pip install 'fluxlit[auth]').

Uses PyJWT with optional jwt.PyJWKClient for JWKS (RS256/ES256) or HS256 for development.

class fluxlit.auth.jwt.JWTAuthConfig(issuer, audience, algorithms=<factory>, jwks_url=None, hs256_secret=None, leeway_seconds=0)[source]¶

Bases: object

Configuration for JWTBearer.

algorithms: list[str]¶

audience: str | list[str]¶

hs256_secret: str | None = None¶: If set, tokens are validated with HS256 using this secret (development).

issuer: str¶

jwks_url: str | None = None¶: JWKS URI (required for asymmetric algorithms unless using HS256-only).

leeway_seconds: int = 0¶

class fluxlit.auth.jwt.JWTBearer(config)[source]¶

Bases: object

FastAPI dependency: validate Authorization: Bearer and return StandardClaims.

classmethod from_fluxlit_settings(settings)[source]¶

Build a bearer dependency from settings / FLUXLIT_JWT_*.

Uses FluxlitSettings. Provide either jwt_hs256_secret (development) or jwt_jwks_url (RS256 / JWKS), plus jwt_issuer and jwt_audience. Error messages name the env vars.

Return type:: JWTBearer

class fluxlit.auth.jwt.RequireRoles(bearer, *roles, roles_claim='roles')[source]¶

Bases: object

Dependency factory: require at least one role from a claim (string or list).

class fluxlit.auth.jwt.RequireScopes(bearer, *scopes, scope_claim='scope')[source]¶

Bases: object

Dependency factory: require OAuth2-style scopes (space-delimited or list claim).

class fluxlit.auth.jwt.StandardClaims(**data)[source]¶

Bases: BaseModel

Common JWT claims plus extra keys from the token payload.

aud: str | list[str] | None¶

exp: int | None¶

iat: int | None¶

iss: str | None¶

jti: str | None¶

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

nbf: int | None¶

scope: str | None¶

sub: str | None¶

fluxlit.auth.jwt.issue_hs256_access_token(*, subject, issuer, audience, secret, ttl_seconds, extra_claims=None)[source]¶

Mint a short-lived HS256 JWT (BFF / dev only — keep secret server-side).

Return type:: str

Generic OIDC client and BFF-style auth routes.

Install fluxlit[auth] for JWT signing used by the BFF exchange.

class fluxlit.auth.oidc.GenericOIDCClient(config)[source]¶

Bases: object

OIDC provider using /.well-known/openid-configuration (Authorization Code + PKCE).

authorization_url(*, redirect_uri, state, code_challenge, scope=None)[source]¶

Return type:: str

exchange_code(*, code, code_verifier, redirect_uri)[source]¶

Return type:: dict[str, Any]

property issuer: str¶

load_discovery_sync()[source]¶

Return type:: None

class fluxlit.auth.oidc.GenericOIDCClientConfig(issuer, client_id, client_secret, http_timeout=30.0)[source]¶

Bases: object

client_id: str¶

client_secret: str¶

http_timeout: float = 30.0¶

issuer: str¶

class fluxlit.auth.oidc.InMemoryOIDCBFFTokenStore(_pkce=<factory>, _exchange=<factory>, _lock=<factory>, state_ttl_seconds=600.0, otc_ttl_seconds=120.0)[source]¶

Bases: object

Process-local PKCE and exchange storage (default for OIDCBFFConfig).

otc_ttl_seconds: float = 120.0¶

pop_exchange_token(auth_code, *, now)[source]¶

Return type:: str | None

pop_pkce_verifier(state, *, now)[source]¶

Return type:: str | None

save_exchange_token(auth_code, access_token, *, now)[source]¶

Return type:: None

save_pkce_verifier(state, code_verifier, *, now)[source]¶

Return type:: None

state_ttl_seconds: float = 600.0¶

class fluxlit.auth.oidc.OIDCBFFConfig(oidc, first_party_secret, token_issuer='fluxlit-bff', token_audience='fluxlit-app', access_token_ttl_seconds=3600, public_base_url='', login_path='/auth/login', callback_path='/auth/callback', exchange_path='/auth/exchange', scope='openid profile email', streamlit_redirect_path='/', state_ttl_seconds=600, otc_ttl_seconds=120, id_token_audience='', id_token_leeway_seconds=0, allow_unverified_id_token_for_custom_oidc=False, bff_token_store=None)[source]¶

Bases: object

Settings for register_oidc_bff_routes().

access_token_ttl_seconds: int = 3600¶

allow_unverified_id_token_for_custom_oidc: bool = False¶

If True, allow non-GenericOIDCClient providers to use parse-only sub.

bff_token_store: OIDCBFFTokenStore | None = None¶

Optional shared storage for PKCE state and one-time auth_code values.

When None, register_oidc_bff_routes() uses InMemoryOIDCBFFTokenStore with state_ttl_seconds and otc_ttl_seconds (single replica).

callback_path: str = '/auth/callback'¶

exchange_path: str = '/auth/exchange'¶

first_party_secret: str¶

id_token_audience: str = ''¶

Expected aud for id_token validation.

If empty, client_id from GenericOIDCClient is used.

id_token_leeway_seconds: int = 0¶: Clock skew leeway (seconds) when validating id_token with JWKS.

login_path: str = '/auth/login'¶

oidc: OIDCProvider¶

otc_ttl_seconds: int = 120¶

public_base_url: str = ''¶: Origin only, e.g. https://app.example.com (no trailing slash).

scope: str = 'openid profile email'¶

state_ttl_seconds: int = 600¶

streamlit_redirect_path: str = '/'¶: Browser path to send the user to with a one-time auth_code query param.

token_audience: str = 'fluxlit-app'¶

token_issuer: str = 'fluxlit-bff'¶

class fluxlit.auth.oidc.OIDCBFFTokenStore(*args, **kwargs)[source]¶

Bases: Protocol

Storage for OIDC BFF PKCE state and one-time auth_code exchange payloads.

Defaults to InMemoryOIDCBFFTokenStore (process memory). Multi-replica API deployments should inject a shared implementation (for example Redis with TTL).

pop_exchange_token(auth_code, *, now)[source]¶

Return type:: str | None

pop_pkce_verifier(state, *, now)[source]¶

Return type:: str | None

save_exchange_token(auth_code, access_token, *, now)[source]¶

Return type:: None

save_pkce_verifier(state, code_verifier, *, now)[source]¶

Return type:: None

class fluxlit.auth.oidc.OIDCProvider(*args, **kwargs)[source]¶

Bases: Protocol

Minimal OIDC surface for discovery and token exchange.

authorization_url(*, redirect_uri, state, code_challenge, scope=None)[source]¶

Return type:: str

exchange_code(*, code, code_verifier, redirect_uri)[source]¶

Return type:: dict[str, Any]

property issuer: str¶

fluxlit.auth.oidc.pkce_pair()[source]¶

Return (code_verifier, code_challenge) for PKCE (S256).

Return type:: tuple[str, str]

fluxlit.auth.oidc.register_oidc_bff_routes(app, config, *, router_prefix='')[source]¶

Attach login, OAuth callback, and Streamlit-friendly token exchange routes.

Return type:: APIRouter

Streamlit helpers for FluxLit BFF auth (query-param auth code exchange).

fluxlit.auth.streamlit.bearer_headers_from_session(st_module, *, session_key='fluxlit_access_token')[source]¶

Build Authorization headers from st.session_state (no logging of secrets).

Return type:: dict[str, str]

fluxlit.auth.streamlit.exchange_auth_code_from_query(st_module, client, *, exchange_path='/auth/exchange', query_key='auth_code', session_key='fluxlit_access_token')[source]¶

If query_key is present, exchange it for an access token and store in session.

Returns the access token when exchanged; otherwise None. On HTTP errors from the exchange endpoint, raises httpx.HTTPStatusError.

Return type:: str | None

fluxlit.auth.streamlit.prepare_streamlit_api_client(st_module, *, exchange_path='/auth/exchange', session_key='fluxlit_access_token', **client_options)[source]¶

One-step Streamlit helper: exchange auth_code (if present), return an ApiClient.

If a bearer token exists in session_key after the exchange (or was already there), returns ApiClient.for_fluxlit() so protected routes work. Otherwise returns an unauthenticated client.

Return type:: ApiClient

URL-bound server session continuity (Phase 2 follow-on — no HTTP cookies).

Use an opaque query parameter (default fluxlit_sid) plus a SessionStore implementation to survive full browser reloads without relying on the browser cookie jar. The gateway already forwards path + query to Streamlit.

Production: replace InMemorySessionStore with a shared store (Redis, etc.) that implements the same protocol. In-memory is single-process only (typical fluxlit dev / one replica).

See the URL session continuity user guide for security (HTTPS, link leakage, TTL) and multipage patterns.

class fluxlit.url_session.InMemorySessionStore(*, max_entries=5000, default_ttl_seconds=86400.0)[source]¶

Bases: object

Process-local SessionStore (dev / single replica).

Not safe across multiple Uvicorn workers or horizontal replicas without a shared backend.

delete(session_id)[source]¶

Return type:: None

get(session_id)[source]¶

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

set(session_id, data, *, ttl_seconds=None)[source]¶

Persist data for session_id; optional time-to-live in seconds.

Return type:: None

class fluxlit.url_session.SessionStore(*args, **kwargs)[source]¶

Bases: Protocol

Server-side persistence for URL-bound session blobs (JSON-serializable dicts).

delete(session_id)[source]¶

Remove session_id from the store if present.

Return type:: None

get(session_id)[source]¶

Return the stored dict or None if missing/expired.

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

set(session_id, data, *, ttl_seconds=None)[source]¶

Persist data for session_id; optional time-to-live in seconds.

Return type:: None

fluxlit.url_session.ensure_url_session(st, store, *, param=None, initial=None, ttl_seconds=None)[source]¶

Ensure st.query_params[param] exists; mint id, seed store, set query param.

Return type:: str

fluxlit.url_session.hydrate_url_session(st, store, *, param=None, merge=True, blob=None)[source]¶

If st.query_params[param] is set, load the store payload into st.session_state.

merge (default): session_state.setdefault(k, v) for each key in the blob so in-flight widget state wins over stale store keys on first paint.
If the param is missing, returns None and does nothing.
If the param is present but the store has no entry, returns the session id without mutating session_state (caller may seed with SessionStore.set()).

Returns the session id string when the query param is present, else None.

Return type:: str | None

fluxlit.url_session.hydrate_url_session_typed(st, store, model, *, param=None, merge=True, strict=False)[source]¶

Like hydrate_url_session(), then validate the store payload as model.

Validates the store blob before merging into st.session_state so invalid payloads do not partially hydrate the UI session.

Returns (session_id_or_none, validated_model_or_none). On validation failure, when strict is false, returns (sid, None); when strict is true, raises pydantic.ValidationError.

Return type:: tuple[str | None, Optional[TypeVar(ModelT, bound= BaseModel)]]

fluxlit.url_session.new_session_id()[source]¶

Return a new URL-safe opaque session identifier (≥128 bits).

Return type:: str

fluxlit.url_session.persist_url_session(st, store, *, param=None, ttl_seconds=None)[source]¶

Write current st.session_state (shallow dict copy) to the store for param id.

Returns the session id if the param is present, else None.

Values are copied best-effort; remote stores should JSON-encode—non-JSON-safe values may need app-side filtering before SessionStore.set().

If snapshotting session_state fails, the store is not updated but the session id is still returned; a warning is logged (with traceback when FLUXLIT_DEBUG=1).

Return type:: str | None

Deep links to Streamlit pages: normalized query params and optional ?page= routing.

fluxlit.deep_links.match_nav_page(params, pages, *, page_key='page')[source]¶

Resolve (path, title) from a page-style query key and registered pages.

Compares the page_key value (default "page") to each page title, path, and Streamlit-style slug ("/" → "home"; otherwise the path without slashes). The first match in pages wins (order matches pages).

Returns None when the key is missing, empty, or unmatched.

Return type:: tuple[str, str] | None

fluxlit.deep_links.query_params(st)[source]¶

Return the current URL query as plain str values (first value if a list).

Streamlit (and AppTest) may expose st.query_params values as strings or lists when multiple values share a key. This helper normalizes to a single str per key so page logic and email-link prefill stay simple.

Missing or unreadable query_params yields an empty dict.

Return type:: dict[str, str]

Security middleware and helpers.

class fluxlit.security.SecurityHeadersMiddleware(app, dispatch=None)[source]¶

Bases: BaseHTTPMiddleware

Add baseline security headers (opt-in via FluxlitSettings).

async dispatch(request, call_next)[source]¶

Return type:: Response

Typing helpers for Streamlit page callables.

class fluxlit.streamlit.page.PageFn(*args, **kwargs)[source]¶

Bases: Protocol

Protocol for functions registered with fluxlit.app.FluxLit.page().

Streamlit passes the streamlit module as st and an ApiClient bound to the internal API base. Handlers may return PageMeta for post-run metadata (breadcrumbs, etc.).

The Typer CLI module (fluxlit.cli) is primarily used via the fluxlit console script; its objects are listed below for completeness.

Typer CLI: fluxlit dev, run, workbench, shutdown, doctor, config, build, new.

The fluxlit setuptools entrypoint calls main(). Commands resolve defaults from fluxlit.config.load_project_config() and FluxlitSettings.

fluxlit.cli.build(output=None, target=<typer.models.ArgumentInfo object>, force=<typer.models.OptionInfo object>)[source]¶

Emit a minimal Dockerfile and .dockerignore for container deployment.

Refuses to overwrite existing files unless --force. Adjust generated files for your layout (dependencies, non-root user, etc.).

Return type:: None

fluxlit.cli.config(target=<typer.models.ArgumentInfo object>, json_output=<typer.models.OptionInfo object>, strict=<typer.models.OptionInfo object>)[source]¶

Print effective FluxLit configuration after precedence rules (env, project file, app).

Shows resolved bind defaults, redacted settings, derived internal API base, and warnings with documentation links. Does not start the server.

Return type:: None

fluxlit.cli.dev(target=<typer.models.ArgumentInfo object>, host=<typer.models.OptionInfo object>, port=<typer.models.OptionInfo object>, log_level=<typer.models.OptionInfo object>, proxy_headers=<typer.models.OptionInfo object>, forwarded_allow_ips=<typer.models.OptionInfo object>, reload=<typer.models.OptionInfo object>, reload_scope=<typer.models.OptionInfo object>, pidfile=None, no_pidfile=<typer.models.OptionInfo object>, workbench=<typer.models.OptionInfo object>, debug=<typer.models.OptionInfo object>)[source]¶

Run the unified stack for local development (Streamlit subprocess + Uvicorn gateway).

Resolves target, bind address, port, and log level from CLI, project file, and FluxlitSettings. See fluxlit.runtime.run_unified().

Return type:: None

fluxlit.cli.doctor(target=<typer.models.ArgumentInfo object>, warnings_only=<typer.models.OptionInfo object>, json_output=<typer.models.OptionInfo object>, verbose=<typer.models.OptionInfo object>, check_pages=<typer.models.OptionInfo object>, strict=<typer.models.OptionInfo object>)[source]¶

Print PASS/WARN/FAIL diagnostics (imports, deps, bind, env).

Exits with code 1 if any check fails, unless --warnings-only is set.

Return type:: None

fluxlit.cli.main()[source]¶

Invoke the root Typer application (console script entrypoint).

Return type:: None

fluxlit.cli.new(name=<typer.models.ArgumentInfo object>, profile=<typer.models.OptionInfo object>)[source]¶

Create <name>/app.py with a sample API route and Streamlit home page.

Return type:: None

fluxlit.cli.pages_manifest(target=<typer.models.OptionInfo object>)[source]¶

Print a JSON page manifest for the resolved FluxLit app (manifest_version 1, stable).

Return type:: None

fluxlit.cli.pages_validate(target=<typer.models.OptionInfo object>, strict=<typer.models.OptionInfo object>)[source]¶

Validate page handlers and manifest JSON-serializability (exit 1 on errors).

Return type:: None

fluxlit.cli.run_cmd(target=<typer.models.ArgumentInfo object>, host=<typer.models.OptionInfo object>, port=<typer.models.OptionInfo object>, log_level=<typer.models.OptionInfo object>, proxy_headers=<typer.models.OptionInfo object>, forwarded_allow_ips=<typer.models.OptionInfo object>, pidfile=None, no_pidfile=<typer.models.OptionInfo object>, workbench=<typer.models.OptionInfo object>, debug=<typer.models.OptionInfo object>)[source]¶

Run the unified stack for production-style use (no Uvicorn reload).

Same resolution rules as dev(); always calls fluxlit.runtime.run_unified() with reload=False.

Return type:: None

fluxlit.cli.shutdown(pidfile=None, force=<typer.models.OptionInfo object>, wait_s=<typer.models.OptionInfo object>)[source]¶

Stop fluxlit dev or fluxlit run using the PID file they write.

Run this from the same working directory (or pass --pidfile / set FLUXLIT_PIDFILE) as the server process. Normal exit removes the PID file; this command also deletes it when the process is already gone.

Return type:: None

fluxlit.cli.workbench_cmd(target=<typer.models.ArgumentInfo object>, host=<typer.models.OptionInfo object>, port=<typer.models.OptionInfo object>, log_level=<typer.models.OptionInfo object>, forwarded_allow_ips=<typer.models.OptionInfo object>, pidfile=None, no_pidfile=<typer.models.OptionInfo object>, debug=<typer.models.OptionInfo object>)[source]¶

Run the unified stack for Posit Workbench / Posit Connect-style path proxies.

Equivalent to fluxlit run with --workbench: Uvicorn proxy_headers is enabled, forwarded_allow_ips defaults follow FluxlitSettings, and a startup banner prints suggested loopback URLs (set FLUXLIT_ROOT_PATH for subpaths).

Return type:: None

The Streamlit entry script fluxlit.streamlit.main is executed by streamlit run with FLUXLIT_APP set. It is not import-safe for autodoc (module-level initialization). See the source file streamlit/main.py in the repository.