etlantic-fastapi API

Status: Available in ETLantic 0.41.0. Dual surface: CP1 control plane (ETLanticAPI, include_router, create_app) is primary; create_reference_app is a non-CP sync demo. Hub: Optional packages API. Adopter guide: Control plane (CP1).

Setup

pip install 'etlantic-fastapi==0.41.0'

import etlantic_fastapi
print(etlantic_fastapi.__version__)

Surfaces

Surface	Entry points	Role
CP1 (primary)	ETLanticAPI, include_router, create_app	Embeddable, authz’d, durable 202 accept + SSE
Reference (non-CP)	create_reference_app	Sync AuthoringService demo only

CP1 is incubation — not production multi-tenant GA (0.43). Do not use FastAPI BackgroundTasks for heavy pipeline work. See ADR-016 and the package README.

Failure modes

Topic	Behavior
Cross-tenant / unknown resource	Opaque 404 after authz (non-enumeration)
In-scope action deny	403
SSE unknown / expired cursor	410 (PMCP410); omit cursor to replay
Ready probe without stores	503 status=not_ready (/health stays 200)
Reference app	Sync only; not a control plane

Public API

etlantic_fastapi

Thin FastAPI adapter: control-plane API (CP1) + non-CP reference app.

Package version is 0.39.0 (CP1 gate-ready).

AuthoringService dataclass

AuthoringService(policy: PolicyContext = PolicyContext(), runtime: PipelineRuntime = PipelineRuntime(), definitions: dict[str, PipelineDefinition] = dict(), jobs: dict[str, RunJob] = dict())

In-memory facade mapping request models to public ETLantic operations.

Not a multi-tenant control plane. HTTP adapters (for example etlantic-fastapi) should bind host policy into PolicyContext.

negotiation

negotiation() -> dict[str, Any]

Return capability negotiation metadata for clients.

Returns:

Type	Description
dict[str, Any]	Capability negotiation payload plus run_model.

Raises:

Type	Description
PermissionError	If catalog is not an allowed action.

catalog

catalog(definition_id: str | None = None) -> dict[str, Any]

Return the authoring catalog, optionally scoped to a definition.

Parameters:

Name	Type	Description	Default
definition_id	str | None	Optional stored definition id to enrich the catalog.	None

Returns:

Type	Description
dict[str, Any]	Catalog document as a dict.

Raises:

Type	Description
PermissionError	If catalog is not allowed.

put_definition

put_definition(definition_id: str, document: dict[str, Any], *, idempotency_key: str | None = None) -> dict[str, Any]

Store a verified pipeline definition document.

Parameters:

Name	Type	Description	Default
definition_id	str	Host-assigned definition id.	required
document	dict[str, Any]	etlantic.pipeline/1 mapping (fingerprint verified).	required
idempotency_key	str | None	Reserved for adapters; ignored in this reference.	None

Returns:

Type	Description
dict[str, Any]	Mapping with id, fingerprint, and definition.

Raises:

Type	Description
PermissionError	If edit is not allowed or policy rejects the definition.
ValueError	On fingerprint / schema verification failure.
TypeError	If document is not a mapping.

get_definition

get_definition(definition_id: str) -> dict[str, Any]

Fetch a stored definition.

Parameters:

Name	Type	Description	Default
definition_id	str	Previously stored id.	required

Returns:

Type	Description
dict[str, Any]	Mapping with id, fingerprint, and definition.

Raises:

Type	Description
KeyError	If definition_id is unknown.

apply_edit

apply_edit(definition_id: str, command: dict[str, Any], *, expected_token: str | None = None) -> dict[str, Any]

Apply an immutable edit to a stored definition.

Parameters:

Name	Type	Description	Default
definition_id	str	Stored definition id.	required
command	dict[str, Any]	EditCommand mapping (op, optional path / payload).	required
expected_token	str | None	Optimistic concurrency token (fingerprint).	None

Returns:

Type	Description
dict[str, Any]	Updated id, fingerprint, concurrency token, and definition.

Raises:

Type	Description
PermissionError	If edit is not allowed or policy rejects.
KeyError	If the definition id is unknown.
ValueError	On concurrency or edit payload failures.

validate

validate(definition_id: str) -> dict[str, Any]

Structurally validate a stored definition for the policy profile.

Parameters:

Name	Type	Description	Default
definition_id	str	Stored definition id.	required

Returns:

Type	Description
dict[str, Any]	Mapping with ok, diagnostics, and fingerprint.

Raises:

Type	Description
PermissionError	If validate is not allowed or policy rejects.
KeyError	If the definition id is unknown.

plan

plan(definition_id: str) -> dict[str, Any]

Plan a stored definition for the policy profile.

Parameters:

Name	Type	Description	Default
definition_id	str	Stored definition id.	required

Returns:

Type	Description
dict[str, Any]	Mapping with ok, diagnostics, and optional plan dict.

Raises:

Type	Description
PermissionError	If plan is not allowed or policy rejects.
KeyError	If the definition id is unknown.

submit_run

submit_run(definition_id: str, *, idempotency_key: str | None = None) -> dict[str, Any]

Run synchronously (reference adapter — not a durable async queue).

Parameters:

Name	Type	Description	Default
definition_id	str	Stored definition id.	required
idempotency_key	str | None	Reserved for adapters; ignored here.	None

Returns:

Type	Description
dict[str, Any]	Job status mapping from job_status.

Raises:

Type	Description
PermissionError	If run is not allowed or policy rejects.
KeyError	If the definition id is unknown.

cancel_run

cancel_run(job_id: str) -> dict[str, Any]

Reference adapter runs are synchronous; cancel is not supported in-flight.

Parameters:

Name	Type	Description	Default
job_id	str	Job id from submit_run.	required

Returns:

Type	Description
dict[str, Any]	Status mapping with cancellable / cancel_supported false.

Raises:

Type	Description
PermissionError	If cancel is not allowed.
KeyError	If the job id is unknown.

job_status

job_status(job_id: str) -> dict[str, Any]

Return status for a synchronous reference job.

Parameters:

Name	Type	Description	Default
job_id	str	Job id from submit_run.	required

Returns:

Type	Description
dict[str, Any]	Status mapping including optional report.

Raises:

Type	Description
PermissionError	If report is not allowed.
KeyError	If the job id is unknown.

PolicyContext dataclass

PolicyContext(tenant: str = 'default', environment: str = 'development', profile: str = 'development', allowed_assets: tuple[str, ...] | None = None, allowed_plugins: tuple[str, ...] | None = None, allowed_actions: tuple[str, ...] = ('catalog', 'validate', 'plan', 'edit', 'run', 'cancel', 'report'))

Host-supplied authority; client payloads cannot expand these grants.

Attributes:

Name	Type	Description
tenant	str	Logical tenant label for the host (reference only).
environment	str	Deployment environment label.
profile	str	Profile name resolved for validate/plan/run.
allowed_assets	tuple[str, ...] | None	When set, definition assets must be in this set.
allowed_plugins	tuple[str, ...] | None	When set, profile allowlist and engine refs must match.
allowed_actions	tuple[str, ...]	Actions the facade may perform.

ETLanticAPI dataclass

ETLanticAPI(authorizer: Authorizer, definitions: DefinitionRepository, submissions: SubmissionStore, events: EventStore, context_factory: ContextFactory, principal_dependency: PrincipalDependency = principal_from_header, profile: Any = 'development', history_store: HistoryStore | None = None, known_observation_ids: set[str] = set(), registry: RegistryProvider | None = None, durable_work: DurableWorkStore | None = None, title: str = 'ETLantic Control Plane', version: str = __version__)

Control-plane API holding injected stores and auth adapters.

Heavy pipeline work is never scheduled via FastAPI BackgroundTasks. Submission acceptance is a durable store commit only; optional pollers may observe accepted jobs without executing them in-request.

Optional registry enables /v1/registry admin routes (CP2). Use :meth:with_registry_definitions or definitions_backend="registry" so /v1/definitions* read/write through registry revisions without changing route paths or operationIds.

Optional durable_work enables /v1/durable/* host routes (CP3) and dual-writes durable accept on submit when fingerprints are present.

with_registry_definitions classmethod

with_registry_definitions(*, authorizer: Authorizer, registry: RegistryProvider, submissions: SubmissionStore, events: EventStore, context_factory: ContextFactory, principal_dependency: PrincipalDependency | None = None, **kwargs: Any) -> ETLanticAPI

Build an API whose /v1/definitions* use registry-backed storage.

LandingWatchSubmitter dataclass

LandingWatchSubmitter(watch_dir: Path, definition_id: str, submit_run: SubmitRun, binding_ref: Mapping[str, Any] = local_files_binding_ref(), tenant_id: str = 'default', workspace_id: str = 'default', poll_interval: float = 0.25, pattern: str = '*.csv', _seen: MutableMapping[str, float] = dict())

Stdlib polling loop that submits durable runs for new files.

Prefer this over optional watchdog so CP1 has zero extra deps.

poll_once

poll_once() -> list[Mapping[str, Any]]

Submit for newly seen files; return accept receipts.

run_until

run_until(stop: Event, *, max_submits: int | None = None) -> list[Mapping[str, Any]]

Poll until stop is set or max_submits receipts collected.

create_app

create_app(api: ETLanticAPI | None = None, *, authorizer: Authorizer | None = None, definitions: DefinitionRepository | None = None, submissions: SubmissionStore | None = None, events: EventStore | None = None, context_factory: ContextFactory | None = None, principal_dependency: PrincipalDependency | None = None, registry: RegistryProvider | None = None, durable_work: DurableWorkStore | None = None, definitions_backend: str | None = None, title: str | None = None, version: str | None = None, install_handlers: bool = True, with_lifespan: bool = True) -> FastAPI

Standalone control-plane app factory.

Optional lifespan wires injected stores onto app.state and verifies readiness. It does not start BackgroundTasks or execute pipelines.

definitions_backend="registry" requires registry and wraps it as :class:RegistryDefinitionRepository for stable /v1/definitions* routes. Default keeps an injected definitions store (callers must pass one when not using the registry backend).

include_router

include_router(app: FastAPI, api: ETLanticAPI, *, prefix: str = '') -> None

Embed the control-plane router without owning host lifecycle.

Does not install lifespan hooks, middleware, or exception handlers. Host applications should register Problem Details handlers and lifespan themselves when desired.

make_principal_from_header

make_principal_from_header(*, header: str = 'X-Principal') -> PrincipalDependency

Build a principal dependency that reads header from the request.

membership_context_factory

membership_context_factory(membership: MembershipMap) -> ContextFactory

Build a context factory that maps principal subjects via membership.

oauth2_oidc_principal_hook

oauth2_oidc_principal_hook(*, token_claims: Mapping[str, Any], subject_claim: str = 'sub', issuer_claim: str = 'iss') -> Principal

Placeholder OAuth2/OIDC claim → Principal mapping for host adapters.

Hosts that validate JWTs (or introspect tokens) via FastAPI security dependencies can call this after claims are verified. This function does not validate signatures or contact an IdP.

principal_dependency_from_callable

principal_dependency_from_callable(resolver: Callable[[Request], Principal]) -> PrincipalDependency

Wrap a host-defined principal resolver as a FastAPI dependency.

principal_from_header

principal_from_header(request: Request) -> Principal

Demo/test principal adapter reading an opaque X-Principal header.

Production hosts should replace this with OAuth2/OIDC or mTLS mapping.

static_context_factory

static_context_factory(*, tenant_id: str, workspace_id: str, environment: str = 'development', security_domain: str = 'default') -> ContextFactory

Fixed-scope context factory for single-tenant demos and unit tests.

assert_path_scope

assert_path_scope(ctx: ControlPlaneContext, *, tenant_id: str | None = None, workspace_id: str | None = None) -> None

Reject path scope that does not match server-derived context (404).

Spoofed path tenants must not disclose existence — map to opaque not_found.

make_context_dependency

make_context_dependency(api: ETLanticAPI) -> Callable[..., ControlPlaneContext]

Build a dependency that constructs server-derived context.

Path tenant/workspace segments are never treated as authority. When a route exposes those path params for routing, callers should verify them against the returned context via :func:assert_path_scope.

control_plane_error_handler

control_plane_error_handler(_request: Request, exc: ControlPlaneError) -> JSONResponse

Map :class:ControlPlaneError to an RFC 7807-shaped JSON body.

install_exception_handlers

install_exception_handlers(app: FastAPI) -> None

Install CP error handlers on a standalone app (not used by include_router).

local_files_binding_ref

local_files_binding_ref(*, root_ref: str = 'landing', root: str = 'inbox', glob: str = '*.csv', mode: str = 'snapshot', provider: str = 'local-files', format: str = 'csv') -> dict[str, Any]

0.38 local-files-style binding reference (paths/refs only, no bytes).

make_testclient_submit_run

make_testclient_submit_run(client: Any, *, principal: str = 'alice') -> SubmitRun

Adapt a FastAPI TestClient into a :data:SubmitRun callable.

create_reference_app

create_reference_app(*, service: AuthoringService | None = None, title: str = 'ETLantic Authoring Reference', version: str = __version__) -> FastAPI

Create a FastAPI app exposing the public authoring/service facade.

This is a non-CP sync demo. It is not the control plane and must not be treated as a multi-tenant production surface.

format_sse_message

format_sse_message(event: ControlPlaneEvent) -> str

Encode one ControlPlaneEvent as an SSE message (id = resume cursor).

sse_streaming_response

sse_streaming_response(events: EventStore, ctx: ControlPlaneContext, run_id: str, *, cursor: str | None, follow: bool = False, headers: Mapping[str, str] | None = None, max_polls: int | None = None, max_duration_s: float | None = None) -> StreamingResponse

Build a text/event-stream response for run events.

api

ETLanticAPI — injectable control-plane FastAPI composition root.

ETLanticAPI dataclass

ETLanticAPI(authorizer: Authorizer, definitions: DefinitionRepository, submissions: SubmissionStore, events: EventStore, context_factory: ContextFactory, principal_dependency: PrincipalDependency = principal_from_header, profile: Any = 'development', history_store: HistoryStore | None = None, known_observation_ids: set[str] = set(), registry: RegistryProvider | None = None, durable_work: DurableWorkStore | None = None, title: str = 'ETLantic Control Plane', version: str = __version__)

Control-plane API holding injected stores and auth adapters.

Heavy pipeline work is never scheduled via FastAPI BackgroundTasks. Submission acceptance is a durable store commit only; optional pollers may observe accepted jobs without executing them in-request.

Optional registry enables /v1/registry admin routes (CP2). Use :meth:with_registry_definitions or definitions_backend="registry" so /v1/definitions* read/write through registry revisions without changing route paths or operationIds.

Optional durable_work enables /v1/durable/* host routes (CP3) and dual-writes durable accept on submit when fingerprints are present.

with_registry_definitions classmethod

with_registry_definitions(*, authorizer: Authorizer, registry: RegistryProvider, submissions: SubmissionStore, events: EventStore, context_factory: ContextFactory, principal_dependency: PrincipalDependency | None = None, **kwargs: Any) -> ETLanticAPI

Build an API whose /v1/definitions* use registry-backed storage.

include_router

include_router(app: FastAPI, api: ETLanticAPI, *, prefix: str = '') -> None

Embed the control-plane router without owning host lifecycle.

Does not install lifespan hooks, middleware, or exception handlers. Host applications should register Problem Details handlers and lifespan themselves when desired.

create_app

create_app(api: ETLanticAPI | None = None, *, authorizer: Authorizer | None = None, definitions: DefinitionRepository | None = None, submissions: SubmissionStore | None = None, events: EventStore | None = None, context_factory: ContextFactory | None = None, principal_dependency: PrincipalDependency | None = None, registry: RegistryProvider | None = None, durable_work: DurableWorkStore | None = None, definitions_backend: str | None = None, title: str | None = None, version: str | None = None, install_handlers: bool = True, with_lifespan: bool = True) -> FastAPI

Standalone control-plane app factory.

Optional lifespan wires injected stores onto app.state and verifies readiness. It does not start BackgroundTasks or execute pipelines.

definitions_backend="registry" requires registry and wraps it as :class:RegistryDefinitionRepository for stable /v1/definitions* routes. Default keeps an injected definitions store (callers must pass one when not using the registry backend).

auth

Authentication adapters for the control-plane FastAPI package.

Apps inject a principal dependency. Path/header tenant claims are never authority — only the server-derived :class:ControlPlaneContext is.

Optional OAuth2/OIDC: install fastapi security dependencies in the host application and pass the resulting principal into :func:principal_dependency_from_callable. This package does not bundle an IdP client; the hook shape is documented for host composition.

principal_from_header

principal_from_header(request: Request) -> Principal

Demo/test principal adapter reading an opaque X-Principal header.

Production hosts should replace this with OAuth2/OIDC or mTLS mapping.

make_principal_from_header

make_principal_from_header(*, header: str = 'X-Principal') -> PrincipalDependency

Build a principal dependency that reads header from the request.

principal_dependency_from_callable

principal_dependency_from_callable(resolver: Callable[[Request], Principal]) -> PrincipalDependency

Wrap a host-defined principal resolver as a FastAPI dependency.

oauth2_oidc_principal_hook

oauth2_oidc_principal_hook(*, token_claims: Mapping[str, Any], subject_claim: str = 'sub', issuer_claim: str = 'iss') -> Principal

Placeholder OAuth2/OIDC claim → Principal mapping for host adapters.

Hosts that validate JWTs (or introspect tokens) via FastAPI security dependencies can call this after claims are verified. This function does not validate signatures or contact an IdP.

membership_context_factory

membership_context_factory(membership: MembershipMap) -> ContextFactory

Build a context factory that maps principal subjects via membership.

static_context_factory

static_context_factory(*, tenant_id: str, workspace_id: str, environment: str = 'development', security_domain: str = 'default') -> ContextFactory

Fixed-scope context factory for single-tenant demos and unit tests.

deps

FastAPI dependencies for server-derived ControlPlaneContext.

make_context_dependency

make_context_dependency(api: ETLanticAPI) -> Callable[..., ControlPlaneContext]

Build a dependency that constructs server-derived context.

Path tenant/workspace segments are never treated as authority. When a route exposes those path params for routing, callers should verify them against the returned context via :func:assert_path_scope.

assert_path_scope

assert_path_scope(ctx: ControlPlaneContext, *, tenant_id: str | None = None, workspace_id: str | None = None) -> None

Reject path scope that does not match server-derived context (404).

Spoofed path tenants must not disclose existence — map to opaque not_found.

errors

Problem Details exception handlers for control-plane errors.

control_plane_error_handler

control_plane_error_handler(_request: Request, exc: ControlPlaneError) -> JSONResponse

Map :class:ControlPlaneError to an RFC 7807-shaped JSON body.

install_exception_handlers

install_exception_handlers(app: FastAPI) -> None

Install CP error handlers on a standalone app (not used by include_router).

landing_sensor

Landing-zone directory watch submitter (039-L) — outside ETLantic core.

Continuous directory watching is a submitter, not a third Extract kind. This module lives in etlantic-fastapi (or examples) and must never move under src/etlantic/. Watchers call the durable submit API with workspace-scoped local-files binding refs and must never embed file contents in plan or submit bodies.

SubmitRun module-attribute

SubmitRun = Callable[[str, Mapping[str, Any], str], Mapping[str, Any]]

(definition_id, payload, idempotency_key) -> accept receipt mapping.

LandingWatchSubmitter dataclass

LandingWatchSubmitter(watch_dir: Path, definition_id: str, submit_run: SubmitRun, binding_ref: Mapping[str, Any] = local_files_binding_ref(), tenant_id: str = 'default', workspace_id: str = 'default', poll_interval: float = 0.25, pattern: str = '*.csv', _seen: MutableMapping[str, float] = dict())

Stdlib polling loop that submits durable runs for new files.

Prefer this over optional watchdog so CP1 has zero extra deps.

poll_once

poll_once() -> list[Mapping[str, Any]]

Submit for newly seen files; return accept receipts.

run_until

run_until(stop: Event, *, max_submits: int | None = None) -> list[Mapping[str, Any]]

Poll until stop is set or max_submits receipts collected.

local_files_binding_ref

local_files_binding_ref(*, root_ref: str = 'landing', root: str = 'inbox', glob: str = '*.csv', mode: str = 'snapshot', provider: str = 'local-files', format: str = 'csv') -> dict[str, Any]

0.38 local-files-style binding reference (paths/refs only, no bytes).

file_identity_ref

file_identity_ref(path: Path, *, watch_root: Path) -> dict[str, Any]

Stable relative identity for a dropped file (never contents).

build_submit_payload

build_submit_payload(*, definition_id: str, binding_ref: Mapping[str, Any], file_ref: Mapping[str, Any] | None = None) -> dict[str, Any]

Durable submit payload with landing binding refs only (no file bytes).

idempotency_key_for_file

idempotency_key_for_file(*, tenant_id: str, workspace_id: str, definition_id: str, file_ref: Mapping[str, Any], principal_subject: str | None = None, operation: str = 'run.submit') -> str

Deterministic scoped key so the same drop does not double-accept.

The store applies the full ADR-016 tuple (tenant, workspace, principal, operation, key); this helper hashes the client-visible key material (optionally including principal/operation).

assert_no_file_bytes

assert_no_file_bytes(document: Mapping[str, Any], *, forbidden: str) -> None

Raise AssertionError if forbidden file contents appear in document.

make_testclient_submit_run

make_testclient_submit_run(client: Any, *, principal: str = 'alice') -> SubmitRun

Adapt a FastAPI TestClient into a :data:SubmitRun callable.

reference

Non-control-plane synchronous AuthoringService demo (not CP1).

This module is intentionally separate from the embeddable control-plane API. It does not provide multi-tenant isolation, durable acceptance, or authorization-before-lookup semantics. Prefer :func:create_app / :func:include_router for CP1 work.

create_reference_app

create_reference_app(*, service: AuthoringService | None = None, title: str = 'ETLantic Authoring Reference', version: str = __version__) -> FastAPI

Create a FastAPI app exposing the public authoring/service facade.

This is a non-CP sync demo. It is not the control plane and must not be treated as a multi-tenant production surface.

routes

Control-plane route builders (definitions, runs, stubs, health).

build_control_plane_router

build_control_plane_router(api: ETLanticAPI) -> APIRouter

Build the CP1 router with stable OpenAPI operationIds.

schemas

Pydantic request/response models for the control-plane HTTP adapter.

RunSubmitBody

Bases: BaseModel

Optional body for run submit; Idempotency-Key header is preferred.

sse

Resumable Server-Sent Events for control-plane run observation (039-E).

History fallback (CP1 / ADR-016): unknown or expired resume cursors fail closed with HTTP 410 Gone and a hint to reconnect without cursor / Last-Event-ID to replay from the beginning. They do not silently skip or invent a mid-stream position.

follow=true is capped (default max polls / duration) so CP1 never blocks unbounded on a sync sleep loop.

Optional WebSocket adapters are experimental and not required for the 0.39 exit gate — they are intentionally omitted here.

event_matches_run

event_matches_run(event: ControlPlaneEvent, run_id: str) -> bool

True when the event envelope is associated with run_id.

Matches an explicit run_id in the payload (and optional host-set attribute). Does not treat acceptance_id / submission_id as run identifiers.

format_sse_message

format_sse_message(event: ControlPlaneEvent) -> str

Encode one ControlPlaneEvent as an SSE message (id = resume cursor).

resolve_resume_cursor

resolve_resume_cursor(*, cursor: str | None, last_event_id: str | None) -> str | None

Prefer explicit cursor query param over Last-Event-ID header.

validate_resume_cursor

validate_resume_cursor(events: EventStore, ctx: ControlPlaneContext, cursor: str | None) -> None

Fail closed with 410 when cursor is unknown in scope.

iter_run_sse

iter_run_sse(events: EventStore, ctx: ControlPlaneContext, run_id: str, *, cursor: str | None, follow: bool = False, poll_interval: float = 0.25, heartbeat_every: int = 4, max_polls: int | None = None, max_duration_s: float | None = None) -> Iterator[str]

Yield SSE frames for run_id after cursor.

When follow is False (default), emit matching history then close. When True, poll for new events with periodic heartbeat comments until max_polls or max_duration_s is reached. CP1 defaults cap follow at :data:DEFAULT_FOLLOW_MAX_POLLS polls and :data:DEFAULT_FOLLOW_MAX_DURATION_S seconds (unbounded follow is rejected by applying these defaults).

sse_streaming_response

sse_streaming_response(events: EventStore, ctx: ControlPlaneContext, run_id: str, *, cursor: str | None, follow: bool = False, headers: Mapping[str, str] | None = None, max_polls: int | None = None, max_duration_s: float | None = None) -> StreamingResponse

Build a text/event-stream response for run events.