latent.prefect.config_loader

Automagic YAML configuration loading for Prefect flows.

Classes

FlowConfig

FlowConfig()

Base for a config_schema, declaring the keys latent itself reads.

The merged config spans layers the flow author does not own: latent scaffold writes mlflow: into flows/global.yaml, and latent.prefect.sampling reads sample_size/sample_seed. A schema covering only the flow's own parameters.yaml is therefore incomplete through no fault of its author, so subclassing this is the supported starting point rather than restating these in every flow.

Functions

flow_exists

flow_exists(flow_name: str) -> bool

Whether flow_name names a real flow, on disk or bundled.

Keyed on the package, not on the YAML it ships: three of the five bundled flow packages declare no parameters at all, and "declares no parameters" is the answer {}, not "no such flow".

get_catalog

get_catalog() -> dict[str, Any]

Get the current catalog configuration.

Returns: Catalog dictionary for the current flow

get_config

get_config() -> dict[str, Any] | BaseModel

Get the current flow configuration.

The return type follows the flow's config_schema: a dict without one, a validated model with one. Callers that want to merge or splat the parameters must go through latent.prefect.params instead — {**base, **get_config()} is a tripwire that stops working the day the flow is typed, while {**base, **params} is stable across that change.

Returns: Parameters dictionary or Pydantic model for the current flow

get_current_run_id

get_current_run_id() -> str | None

The active flow run id, or None outside a flow.

Unlike :func:get_flow_name this never raises: every caller (report stamping, report discovery) has a defined meaning for "no run id" — the pre-run-isolation behaviour — and must not be forced into a try/except to reach it.

get_flow_name

get_flow_name() -> str

Get the current flow name.

Returns: Name of the current flow

load_flow_config

load_flow_config(flow_name: str, config_schema: type[T] | None = None, overrides: dict[str, Any] | None = None, arguments: dict[str, Any] | None = None) -> dict[str, Any]

Load parameters.yaml and catalog.yaml for a flow.

Merge priority (highest wins): 1. Caller overrides and arguments (CLI kwargs, direct-call kwargs) 2. User's flow-specific parameters.yaml 3. Global parameters.yaml 4. Bundled library defaults (shipped with the flow package)

Environment variables are substituted into the YAML layers before overrides are applied, so a caller-supplied $-string reaches params verbatim.

Args: flow_name: Name of the flow (e.g., "conversation_simulation") config_schema: Optional Pydantic model to validate parameters against overrides: Optional top-layer parameter overrides, validated with the rest arguments: Optional top-layer values the flow function takes as arguments — payload rather than configuration, so config_schema is not required to declare them (see :func:_validate_params)

Returns: Dict with merged configuration including 'parameters' and 'catalog' keys

reset_current_run_id

reset_current_run_id(token: Token[str | None]) -> None

set_current_config

set_current_config(config: dict[str, Any])

Set the current flow configuration (for use by decorator).

Args: config: Configuration dictionary from load_flow_config

set_current_run_id

set_current_run_id(run_id: str) -> Token[str | None]

Bind the active flow run id; pass the token to :func:reset_current_run_id.

Token-scoped rather than assigned, so a subflow restores its parent's id on exit and a finished run leaves nothing behind for the next one to inherit.