Skip to content

API Reference

This page is generated from the library docstrings via mkdocstrings.

Core spec

spec

Configuration schema and validation for signac-driven pipelines.

ConfigValidationError

Bases: Exception

Raised when a configuration file fails validation.

DependencySpec dataclass

Represents a single parent dependency for an action.

Attributes

action Name of the parent action. sp_key State point key used to store the parent job id in the child job.

ActionSpec dataclass

Describes an action (stage) in the workflow.

dep_sp_key property 

dep_sp_key: str

SP key used to store this action's parent job ID.

from_mapping staticmethod 

from_mapping(data: Dict[str, Any]) -> 'ActionSpec'

Create an :class:ActionSpec from a mapping.

Parameters

data Mapping parsed from config (TOML/YAML). Must contain name and may provide sp_keys, outputs, runner, and a deps table with action and optional sp_key.

Returns

ActionSpec Parsed action definition with optional dependency.

Raises

ConfigValidationError If required fields are missing or have an unexpected structure.

WorkspaceSpec dataclass

Metadata about the workspace configuration.

from_mapping staticmethod 

from_mapping(data: Dict[str, Any]) -> 'WorkspaceSpec'

Create a workspace spec from a mapping.

Parameters

data Mapping parsed from config; if empty defaults are used.

Returns

WorkspaceSpec Workspace settings (currently only value_file).

WorkflowSpec

Parsed configuration for a workflow.

Provides validation, topological ordering, and access to experiments.

experiments property 

experiments: List[Dict[str, Dict[str, Any]]]

List of experiment parameter blocks copied from the config.

load staticmethod 

load(path: str | Path) -> 'WorkflowSpec'

Load and validate a workflow spec from a TOML or YAML file.

from_mapping staticmethod 

from_mapping(data: Dict[str, Any]) -> 'WorkflowSpec'

Construct a spec from a parsed mapping (already loaded TOML/YAML).

topological_actions 

topological_actions() -> List[ActionSpec]

Return actions ordered so parents come before children.

Raises

ConfigValidationError If the dependency graph contains a cycle.

get_action 

get_action(name: str) -> ActionSpec

Return an action by name or raise :class:ConfigValidationError.

load_spec 

load_spec(path: str | Path) -> WorkflowSpec

Load and validate a workflow spec from disk (TOML or YAML).

Context

context

Context binding of a workflow spec to a signac project.

WorkflowContext

A validated workflow spec bound to a signac project.

materialize 

materialize(
    experiments=None, dry_run: bool = False
) -> MaterializationReport

Create jobs for the provided experiments (or those in the spec).

Materialization

materialize

Job materialization from a workflow spec and experiments.

MaterializationReport

Bases: Struct

Summary of materialization results.

Materializer

Materialize jobs for a workflow in a readable, stepwise flow.

run 

run(
    experiments: Iterable[
        Mapping[str, Mapping[str, object]]
    ],
) -> MaterializationReport

Create jobs for all actions across all experiments.

Parameters

experiments List of per-action parameter mappings, typically spec.experiments.

Returns

MaterializationReport Counts of created/total jobs and ids grouped by action.

Raises

ConfigValidationError If experiments include unknown parameters or are missing required parents.

materialize 

materialize(
    spec: WorkflowSpec,
    project: Project,
    experiments: Iterable[
        Mapping[str, Mapping[str, object]]
    ],
    dry_run: bool = False,
) -> MaterializationReport

Create jobs for all actions across all experiments.

Parameters

spec Validated workflow specification. project signac project to bind jobs to. experiments List of per-action parameter mappings, typically spec.experiments. dry_run If True, compute ids without writing to disk or touching docs.

Returns

MaterializationReport Counts of created/total jobs and ids grouped by action.

Raises

ConfigValidationError If experiments include unknown parameters or are missing required parents.

Helpers

helpers

Convenience helpers for accessing parent jobs and their files.

DependencyResolutionError

Bases: Exception

Raised when a job's parent cannot be resolved.

write_dependency_metadata 

write_dependency_metadata(
    job: Job, parent_job: Job
) -> None

Update job.doc['deps_meta'] with parent job info.

get_parent 

get_parent(
    job: Job, sp_key: str = DEFAULT_PARENT_SP_KEY
) -> Job

Return the parent job referenced by sp_key in the state point.

Parameters

job The child job whose parent should be resolved. sp_key State-point key that holds the parent job ID. Defaults to DEFAULT_PARENT_SP_KEY ("parent_action"); pass action.dep_sp_key when using a custom dependency key.

Raises

DependencyResolutionError If the job does not declare a parent or the referenced job cannot be opened.

open_parent_folder 

open_parent_folder(
    job: Job, path: str | Path | None = None
) -> Path

Return the filesystem path to the parent job workspace, optionally joined with path.

parent_file 

parent_file(
    job: Job, relpath: str | Path, must_exist: bool = True
) -> Path

Return an absolute path to a file in the parent workspace.

parent_product_exists 

parent_product_exists(
    job: Job, relpath: str | Path
) -> bool

Return True if a given file exists in the parent workspace.

parent_path 

parent_path(job: Job) -> Path

Return the parent workspace path for a job.

iter_parent_products 

iter_parent_products(
    job: Job, pattern: str = "*"
) -> Iterator[Path]

Yield paths matching a glob pattern inside the parent workspace.

open_job_from_directory 

open_job_from_directory(directory: str) -> Job

Open a job by workspace directory name (as passed by row).

get_parent_doc 

get_parent_doc(
    job: Job, key: str, default: Any = None
) -> Any

Return a value from the parent job document, ignoring reserved keys.

Typed parameters (runtime-only)

bindings

Registry mapping action names to Pydantic model classes.

WorkflowBindings

Registry that maps action names to Pydantic :class:~pydantic.BaseModel classes.

Usage::

bindings = (
    WorkflowBindings()
    .bind("train", TrainParams)
    .bind("evaluate", EvalParams)
)

bind 

bind(
    action_name: str, model_cls: type[BaseModel]
) -> "WorkflowBindings"

Register model_cls as the schema for action_name.

Parameters

action_name Workflow action identifier (must match the action key in the signac state point). model_cls A Pydantic :class:~pydantic.BaseModel subclass.

Returns

WorkflowBindings The same registry instance, for method chaining.

Raises

ValueError If action_name already has a registered binding. TypeError If model_cls is not a Pydantic BaseModel subclass.

get 

get(action_name: str) -> type[BaseModel]

Return the model class registered for action_name.

Raises

UnknownActionBindingError If no binding exists for action_name.

has 

has(action_name: str) -> bool

Return True if action_name has a registered binding.

runtime

Runtime helpers for loading and validating typed action parameters.

load_action_params 

load_action_params(
    job: Any,
    action_or_model: str
    | type[BaseModel]
    | WorkflowBindings,
    bindings: WorkflowBindings | None = None,
) -> Any

Load and validate the parameters for an action from job.

Three calling styles are supported:

.. code-block:: python

# 1. Original — explicit action name + bindings registry
params = load_action_params(job, "train", bindings)

# 2. Inferred action — pass the registry, action read from job.sp["action"]
params = load_action_params(job, bindings)

# 3. Direct model class — no registry needed, action read from job.sp["action"]
params = load_action_params(job, TrainParams)

Steps (all forms):

  1. Resolve the action name and Pydantic model class.
  2. Extract the raw parameter mapping from job via :func:_extract_action_raw_params.
  3. Validate the mapping with model_cls.model_validate(raw).
  4. Return the validated model instance.
Parameters

job A signac :class:~signac.Job instance whose state point contains the parameters for the action. action_or_model One of:

* ``str`` — explicit action name (original API); *bindings* must also
  be provided.
* :class:`WorkflowBindings` — registry to look up the model; the
  action name is inferred from ``job.sp["action"]``.
* ``type[BaseModel]`` subclass — used directly as the model class; the
  action name is inferred from ``job.sp["action"]``.

bindings :class:WorkflowBindings registry. Required only when action_or_model is a str; ignored otherwise.

Returns

pydantic.BaseModel A validated instance of the resolved model class.

Raises

TypeError If action_or_model is a str but bindings is None, or if an unsupported type is passed for action_or_model. UnknownActionBindingError If action_or_model is a str and bindings has no entry for it. ActionParamsNotFoundError If job does not belong to the resolved action, or if the action cannot be inferred (no action key in the state point). TypedParamsValidationError If Pydantic validation of the raw params fails.

dump_action_params 

dump_action_params(model: BaseModel) -> dict[str, Any]

Serialise a validated Pydantic model back to a plain :class:dict.

Thin wrapper around :meth:pydantic.BaseModel.model_dump.

Parameters

model A Pydantic model instance previously returned by :func:load_action_params.

Returns

dict[str, Any] Plain dictionary representation of the model.

errors

Exceptions raised by the grubicy.typed runtime helpers.

UnknownActionBindingError

Bases: Exception

Raised when no Pydantic model is registered for the requested action.

ActionParamsNotFoundError

Bases: Exception

Raised when a job does not contain parameters for the requested action.

TypedParamsValidationError

Bases: Exception

Raised when Pydantic validation fails for action parameters.

The exception message lists every validation error prefixed with <action>.<field_path> so failures are easy to locate.

Attributes

action Name of the action whose params failed validation. validation_errors Raw list of Pydantic error dicts (from ValidationError.errors()).

Row workflow rendering

row_render

Render a row workflow configuration from a workflow spec.

render_row_workflow 

render_row_workflow(
    spec: WorkflowSpec,
    output_path: str | Path,
    default_command: Optional[str] = None,
) -> Path

Render a row workflow TOML file for the provided spec.

Parameters

spec Validated workflow specification. output_path Where to write the workflow TOML. default_command Optional format string used when an action does not define runner. Defaults to python actions/{name}.py {directory}.

Collection

collect

Utilities to collect flattened parameters and docs across parent chains.

CollectedRow

Bases: Struct

A flattened view of a job and its ancestor parameters/docs.

ParamCollector

Collect flattened parameter/doc rows for a target action.

collect_params_with_parents 

collect_params_with_parents(
    spec: WorkflowSpec,
    project: Project,
    target_action: str,
    *,
    include_doc: bool = False,
    missing_ok: bool = False,
) -> List[CollectedRow]

Collect rows for jobs of target_action with flattened parent params/docs.

Parameters

spec Workflow specification (used to derive the dependency chain and sp_keys). project signac project containing jobs. target_action Action name whose jobs should be collected. include_doc If True, include non-reserved document keys for each action as <action>.doc.<key>. missing_ok If True, skip rows with missing parent jobs; otherwise raise.

Migration

migrate

Schema migrations with cascading dependency rewrites.

MigrationCollisionError

Bases: Exception

Raised when multiple jobs converge to the same new state point.

MigrationLockError

Bases: Exception

Raised when a migration lock cannot be acquired.

MigrationPlan

Bases: Struct

to_json 

to_json() -> str

Return the plan as a JSON string.

MigrationReport

Bases: Struct

to_json 

to_json() -> str

Return the report as a JSON string.

MigrationExecutor

Execute a migration plan and cascade dependency pointer rewrites.

plan_migration 

plan_migration(
    spec: WorkflowSpec,
    project: Project,
    action_name: str,
    sp_transform: Callable[[StatePoint], StatePoint],
    selection: Optional[Mapping[str, Any]] = None,
    collision_strategy: str = "abort",
    plan_path: Optional[Path] = None,
) -> tuple[MigrationPlan, Path]

Create a migration plan for a single action.

The plan computes old->new state points and detects collisions before any mutation.

execute_migration 

execute_migration(
    spec: WorkflowSpec,
    project: Project,
    plan: MigrationPlan,
    plan_path: Optional[Path] = None,
    resume: bool = True,
) -> MigrationReport

Execute a migration plan and cascade dependency pointer rewrites.

Writes progress under .pipeline_migrations so a rerun can resume safely.