Public Python API

Generated from the reviewed core sources using scripts/update_reference.py. Builds consume this checked-in reference and do not import or discover plugins. Version and tag provenance are recorded in the version matrix.

Use plugin contracts to obtain tool configurations and workflows. Call submit_workflow for backend-aware execution; direct workflow calls are intended for tests and custom inline execution. See the first workflow.

Dataset.from_file() currently raises NotImplementedError. A frozen dataclass prevents field reassignment; it does not make nested molecule objects or lists immutable.

caddforge.config.base

BaseConfig

class BaseConfig(BaseModel)

Common config options shared across tools.

Declared fields and enum values (inherited fields are documented on the base class):

run_directory: Path | None = None
registry_path: Path | None = None
backend: str = 'local'
slurm_partition: str | None = None
slurm_time: str | None = None
slurm_account: str | None = None
sbatch_options: list[str] = Field(default_factory=list)
slurm_setup_commands: list[str] = Field(default_factory=list)
job_name: str | None = None
export_results: bool = False
slurm_nodes: int = 1
slurm_ntasks: int = 1
slurm_cpus_per_task: int = 1
slurm_env_pack_path: Path | None = None
slurm_env_cache_enabled: bool = True
slurm_env_cache_dir: Path | None = None
slurm_wheelhouse_path: Path | None = None
slurm_requirements_path: Path | None = None

from_yaml

def from_yaml(cls, path: str | Path) -> 'BaseConfig'

Load configuration from a YAML file.

Args:
    path: Path to the YAML config file.

Returns:
    BaseConfig: Parsed and validated configuration.

from_json

def from_json(cls, path: str | Path) -> 'BaseConfig'

Load configuration from a JSON file.

Args:
    path: Path to the JSON config file.

Returns:
    BaseConfig: Parsed and validated configuration.

example_dict

def example_dict(cls) -> dict[str, Any]

Return a dict with defaults or placeholders for required fields.

Returns:
    dict[str, Any]: Example config mapping.

caddforge.core.dataset

Dataset

class Dataset()

Immutable dataset consisting of entries and optional labels.

Args:
    entries: Sequence of entries (typically molecule-like objects).
    labels: Optional sequence of labels aligned with entries.
    source: Optional provenance string (e.g., file path or database tag).

Declared fields and enum values (inherited fields are documented on the base class):

entries: Sequence[object]
labels: Sequence[object] | None = None
source: str | None = None

from_file

def from_file(cls, path: str | Path) -> 'Dataset'

Create a dataset from a file path.

Args:
    path: Path to the dataset file.

Returns:
    Dataset: Parsed dataset instance.

Raises:
    NotImplementedError: Always, until a concrete loader is implemented.

from_molecules

def from_molecules(cls, molecules: Iterable[object], *, source: str | None=None) -> 'Dataset'

Build a dataset from an iterable of molecule-like objects.

Args:
    molecules: Iterable of molecule-like objects.
    source: Optional provenance string.

Returns:
    Dataset: New dataset with entries populated and labels empty.

MoleculeSet

class MoleculeSet()

Immutable collection of molecules.

Args:
    molecules: Sequence of molecule-like objects.
    source: Optional provenance string (e.g., file path or database tag).

Declared fields and enum values (inherited fields are documented on the base class):

molecules: Sequence[object]
source: str | None = None

caddforge.core.model

Model

class Model(ABC)

Abstract base class for models used across tools.

fit

def fit(self, dataset: Dataset) -> None

Train the model with the provided dataset.

Args:
    dataset: Dataset containing training data.

predict

def predict(self, data: Dataset | MoleculeSet)

Generate predictions for a dataset or molecule set.

Args:
    data: Dataset or MoleculeSet to predict on.

Returns:
    Any: Tool-specific prediction output.

caddforge.core.job

JobStatus

class JobStatus(str, Enum)

Enumeration of job lifecycle states.

Declared fields and enum values (inherited fields are documented on the base class):

PENDING = 'PENDING'
RUNNING = 'RUNNING'
COMPLETED = 'COMPLETED'
FAILED = 'FAILED'
CANCELLED = 'CANCELLED'

Job

class Job()

Canonical representation of a workflow execution.

Args:
    id: Unique job identifier.
    tool_name: Registered tool name (without section).
    backend: Backend name used to execute the job.
    status: Current job status.
    created_at: Timestamp when the job was created.
    started_at: Timestamp when execution began.
    finished_at: Timestamp when execution completed.
    run_directory: Path to the job run directory.
    metadata: Tool/backend-specific metadata.

Declared fields and enum values (inherited fields are documented on the base class):

id: str
tool_name: str
backend: str = 'unsubmitted'
status: JobStatus = JobStatus.PENDING
created_at: datetime = field(default_factory=lambda: datetime.now().astimezone())
started_at: Optional[datetime] = None
finished_at: Optional[datetime] = None
run_directory: Optional[Path] = None
metadata: dict[str, Any] = field(default_factory=dict)

to_metadata_dict

def to_metadata_dict(self) -> dict[str, Any]

Serialize job metadata to a JSON-friendly dict.

Returns:
    dict[str, Any]: Serialized metadata suitable for JSON encoding.

from_metadata_dict

def from_metadata_dict(cls, data: dict[str, Any]) -> 'Job'

Deserialize job metadata produced by `to_metadata_dict`.

Args:
    data: Serialized job metadata.

Returns:
    Job: Deserialized Job instance.

save_metadata

def save_metadata(self, path: Path) -> None

Persist job metadata to disk.

Args:
    path: Target metadata JSON file path.

caddforge.core.result

Result

class Result()

Standardized result object produced by workflows.

Args:
    job_id: Job identifier.
    tool_name: Tool name that produced the result.
    status: Completion status for the job.
    metrics: Tool-defined metrics (simple scalar values).
    paths: Paths to key output artifacts.
    extra: Tool-defined extra metadata payload.

Declared fields and enum values (inherited fields are documented on the base class):

job_id: str
tool_name: str
status: JobStatus
metrics: dict[str, float | int | str] = field(default_factory=dict)
paths: dict[str, Path] = field(default_factory=dict)
extra: dict[str, Any] = field(default_factory=dict)

to_dict

def to_dict(self) -> dict[str, Any]

Serialize to a JSON-friendly dictionary.

Returns:
    dict[str, Any]: Serialized representation of the result.

from_dict

def from_dict(cls, data: dict[str, Any]) -> 'Result'

Construct a Result from a dictionary representation.

Args:
    data: Serialized result dictionary.

Returns:
    Result: Deserialized Result instance.

save

def save(self, path: Path) -> None

Persist result to disk as JSON.

Args:
    path: Target JSON file path.

caddforge.core.workflow

Workflow

class Workflow(ABC)

Abstract workflow run by CADDForge tools.

Declared fields and enum values (inherited fields are documented on the base class):

tool_name: str

prepare_job

def prepare_job(self) -> Job

Build a job object with metadata and paths.

Returns:
    Job: Prepared job instance ready for submission.

run

def run(self, job: Job, context: RuntimeContext | None=None) -> Result

Execute the workflow logic and return a result.

Args:
    job: Job metadata for the execution.
    context: Backend-neutral runtime details supplied by the executor.

Returns:
    Result: Standardized workflow result.

caddforge.core.execution

BackendOptions

class BackendOptions()

Backend-specific options supplied by CLI or GUI callers.

Declared fields and enum values (inherited fields are documented on the base class):

slurm_partition: str | None = None
slurm_time: str | None = None
slurm_account: str | None = None
slurm_nodes: int | None = None
slurm_ntasks: int | None = None
slurm_cpus_per_task: int | None = None
slurm_setup: list[str] = field(default_factory=list)
slurm_env_pack: Path | None = None
slurm_wheelhouse: Path | None = None
slurm_requirements: Path | None = None
extra_sbatch_options: list[str] = field(default_factory=list)

submit_workflow

def submit_workflow(workflow, cfg: BaseConfig, *, backend_name: str, runs_root: Path | str=Path('runs'), registry_path: Path | None=None, job_name: str | None=None, submission_cwd: Path | None=None, local_async: bool=False, backend_options: BackendOptions | None=None, emit: EmitFn | None=None, prepared_job_callback: Callable[[Job], None] | None=None, input_ports: tuple[InputPort, ...]=()) -> Job

Prepare a workflow job and submit it through the selected backend.

load_result_for_job

def load_result_for_job(job: Job) -> Result | None

Load the canonical result for a job if it exists.

caddforge.core.backend

Backend

class Backend(ABC)

Abstract compute backend.

Declared fields and enum values (inherited fields are documented on the base class):

name: str

submit

def submit(self, job: Job, run_callable: Callable[[Job], Result]) -> JobHandle

Submit a job for execution using the provided callable.

Args:
    job: Job metadata to register and execute.
    run_callable: Callable that performs the work and returns a Result.

Returns:
    JobHandle: Backend-specific handle for tracking execution.

submit_shell_commands

def submit_shell_commands(self, job: Job, commands: Iterable[str]) -> JobHandle

Submit a shell-command job if the backend supports it.

Args:
    job: Job metadata to register and execute.
    commands: Shell commands to execute in order.

Returns:
    JobHandle: Backend-specific handle for tracking execution.

Raises:
    NotImplementedError: If the backend does not implement shell jobs.

query_status

def query_status(self, handle: JobHandle) -> JobStatus

Return current status of a submitted job.

Args:
    handle: Backend-specific job handle.

Returns:
    JobStatus: Current status for the job.

cancel

def cancel(self, handle: JobHandle) -> None

Cancel a submitted job if supported.

Args:
    handle: Backend-specific job handle.

caddforge.core.artifacts

ArtifactKind

class ArtifactKind(str, Enum)

Portable artifact categories understood by the core pipeline layer.

Declared fields and enum values (inherited fields are documented on the base class):

FILE = 'file'
DIRECTORY = 'directory'
MOLECULE_COLLECTION = 'molecule_collection'
POSE_COLLECTION = 'pose_collection'
POSE_MANIFEST = 'pose_manifest'
TABULAR_SCORES = 'tabular_scores'
PROTEIN_STRUCTURE = 'protein_structure'
COMPLEX_STRUCTURE = 'complex_structure'

ArtifactRef

class ArtifactRef()

Reference to a workflow artifact stored on disk.

Large data stays on disk; the pipeline engine passes these lightweight
references between stages and resolves them into concrete config paths.

Declared fields and enum values (inherited fields are documented on the base class):

id: str
kind: ArtifactKind | str
producer_stage_id: str
path: Path | None = None
paths: list[Path] = field(default_factory=list)
producer_job_id: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)

primary_path

def primary_path(self) -> Path

Return the single concrete path for config binding.

to_dict

def to_dict(self) -> dict[str, Any]

Serialize to a JSON-friendly dictionary.

from_dict

def from_dict(cls, data: dict[str, Any]) -> ArtifactRef

Deserialize an artifact reference from a dictionary.

InputPort

class InputPort()

Declarative input accepted by a tool plugin.

``config_field`` is the stable top-level pipeline binding target.
``runtime_selector`` may identify nested path values for execution-time
staging without changing that pipeline contract.

Declared fields and enum values (inherited fields are documented on the base class):

name: str
kind: ArtifactKind | str
config_field: str
required: bool = True
description: str | None = None
compression: InputCompressionPolicy | str = InputCompressionPolicy.TRANSPARENT_GZIP
runtime_selector: str | None = None

OutputPort

class OutputPort()

Declarative output produced by a tool plugin.

``config_field`` identifies the optional Path-typed config field that asks
the workflow where to write this result. Declaring it enables transparent
gzip output handling at the pipeline boundary.

Declared fields and enum values (inherited fields are documented on the base class):

name: str
kind: ArtifactKind | str
result_path_key: str
description: str | None = None
required: bool = True
path_collection_column: str | None = None
config_field: str | None = None

SplitContract

class SplitContract()

Central description of a splittable workflow input.

Declared fields and enum values (inherited fields are documented on the base class):

input_field: str
kind: ArtifactKind | str = ArtifactKind.MOLECULE_COLLECTION
strategy: Literal['sdf', 'csv', 'tsv', 'lines', 'auto'] = 'auto'
merge_outputs: dict[str, Literal['csv', 'tsv', 'sdf', 'jsonl', 'manifest', 'auto']] = field(default_factory=dict)
item_path_field: str | None = None

caddforge.core.tool

ToolPlugin

class ToolPlugin(Protocol)

Protocol that tool plugins should satisfy.

Attributes:
    section: Section/group name for CLI organization.
    name: Tool name within its section.
    config_cls: Pydantic config model used by the tool.
    workflow_cls: Workflow class executed by the tool.

Declared fields and enum values (inherited fields are documented on the base class):

section: str
name: str
config_cls: Type[BaseConfig]
workflow_cls: Type[Workflow]
input_ports: tuple[InputPort, ...] = ()
output_ports: tuple[OutputPort, ...] = ()
split_contracts: tuple[SplitContract, ...] = ()

ToolSpec

class ToolSpec()

Concrete description of a discovered tool plugin.

Args:
    section: Section/group name for CLI organization.
    name: Tool name within its section.
    config_cls: Pydantic config model used by the tool.
    workflow_cls: Workflow class executed by the tool.
    plugin: Loaded plugin instance.

Declared fields and enum values (inherited fields are documented on the base class):

section: str
name: str
config_cls: Type[BaseConfig]
workflow_cls: Type[Workflow]
plugin: ToolPlugin

caddforge.plugins

get_tool

def get_tool(section: str, name: str) -> ToolSpec

Return the tool spec for a given section/tool name.

Args:
    section: Section/group name.
    name: Tool name within the section.

Returns:
    ToolSpec: Discovered tool specification.

Raises:
    KeyError: If the section or tool name is not registered.

caddforge.pipeline.spec

PipelineSpec

class PipelineSpec(BaseModel)

Top-level declarative pipeline specification.

Declared fields and enum values (inherited fields are documented on the base class):

version: int = 1
name: str = 'pipeline'
defaults: PipelineDefaults = Field(default_factory=PipelineDefaults)
stages: list[PipelineStageSpec]

PipelineStageSpec

class PipelineStageSpec(BaseModel)

One stage in a CADDForge pipeline DAG.

Declared fields and enum values (inherited fields are documented on the base class):

id: str
tool: str | None = None
type: str | None = None
config: str | Path | dict[str, Any] | None = None
inputs: dict[str, str] = Field(default_factory=dict)
outputs: dict[str, str | dict[str, Any]] = Field(default_factory=dict)
scatter: ScatterSpec | None = None
depends_on: list[str] = Field(default_factory=list)

PipelineDefaults

class PipelineDefaults(BaseModel)

Defaults applied to tool stages.

Declared fields and enum values (inherited fields are documented on the base class):

backend: str = 'local'
run_directory: Path | None = None
registry_path: Path | None = None
job_name: str | None = None
export_results: bool = False

ScatterSpec

class ScatterSpec(BaseModel)

Fan-out settings for a splittable stage.

Declared fields and enum values (inherited fields are documented on the base class):

by: str
chunks: int | None = Field(default=None, ge=1)
chunk_size: int | None = Field(default=None, ge=1)
strategy: Literal['auto', 'sdf', 'csv', 'tsv', 'lines'] = 'auto'

load_pipeline_spec

def load_pipeline_spec(path: str | Path) -> PipelineSpec

Load a pipeline spec from YAML or JSON.

caddforge.pipeline.executor

PipelineExecutionResult

class PipelineExecutionResult()

Summary returned after a pipeline run.

Declared fields and enum values (inherited fields are documented on the base class):

pipeline_job: Job
stage_jobs: dict[str, list[Job]] = field(default_factory=dict)
artifacts: dict[str, ArtifactRef] = field(default_factory=dict)

run_pipeline

def run_pipeline(path: str | Path, *, emit: EmitFn | None=None, poll_interval_seconds: int=10, wait_timeout_seconds: int | None=None, backend_options: BackendOptions | None=None) -> PipelineExecutionResult

Load and run a pipeline spec from disk.