Pipeline¶

Client for managing pipelines with data fitting, calculations, and validation.

For usage examples and guides, see Simulate API on docs.ionworks.com.

Pipeline client for running parameterization workflows.

This module provides the PipelineClient for creating and managing pipelines that combine data fitting, calculations, and validation steps for battery model parameterization.

Pipeline shape validation is delegated to ionworks.pipeline — both this client’s PipelineClient.create() method and the backend route parse against the same schema, so a payload that builds with iw.pipeline.Pipeline(...) validates identically end-to-end.

class ionworks.api.pipeline.PipelineSubmissionMetadata(*, project_id: str | None = None, options: dict[str, Any] | None = None)[source]¶

Bases: BaseModel

SDK-only metadata attached to a pipeline submission.

These fields are not part of ionworks.pipeline.Pipeline because they describe how the submission is routed (which project, which runtime options) rather than what the pipeline does.

project_id: str | None¶

options: dict[str, Any] | None¶

resolve_project_id() → PipelineSubmissionMetadata[source]¶

Resolve project_id from env vars if not provided.

Prefers IONWORKS_PROJECT_ID; falls back to the deprecated PROJECT_ID (with a DeprecationWarning) via resolve_env_project_id().

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

class ionworks.api.pipeline.DataFitResponse(*, parameter_values: dict[str, Any])[source]¶

Bases: BaseModel

Response from a data fitting step containing fitted parameters.

parameter_values: dict[str, Any]¶

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

class ionworks.api.pipeline.CalculationResponse(*, parameter_values: dict[str, Any])[source]¶

Bases: BaseModel

Response from a calculation step containing calculated parameters.

parameter_values: dict[str, Any]¶

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

class ionworks.api.pipeline.EntryResponse(*, parameter_values: dict[str, Any])[source]¶

Bases: BaseModel

Response from an entry point containing parameter values.

parameter_values: dict[str, Any]¶

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

class ionworks.api.pipeline.PipelineSubmissionResponse(*, url: str | None = None, api_url: str | None = None, id: str, name: str, description: str | None = None, project_id: str | None = None, status: str, error: str | None = None, is_terminal: bool, is_failed: bool)[source]¶

Bases: ResourceLinks

Response from submitting a pipeline to the API.

id: str¶

name: str¶

description: str | None¶

project_id: str | None¶: The project the pipeline runs in.

status: str¶

error: str | None¶

is_terminal: bool¶: Server-derived, so the client never hard-codes the status taxonomy.

is_failed: bool¶

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

class ionworks.api.pipeline.PipelineResponse(*, result: dict[str, Any], element_results: dict[str, Any])[source]¶

Bases: BaseModel

Complete response from retrieving pipeline results.

result and element_results are the raw dicts returned by the backend, unchanged. element() and results additionally decode those raw per-element dicts into typed ionworks.pipeline result objects (lazily, cached), wiring up overlay/trace/posterior fetchers per element.

result: dict[str, Any]¶

element_results: dict[str, Any]¶

__init__(**data: Any) → None[source]¶

Build the response, accepting the back-refs as keyword arguments.

Pydantic silently drops underscore-prefixed keyword arguments, so client / pipeline_id are popped here and assigned to their private attributes. Passing them at construction keeps a response from ever existing without them — the lazy fetchers no-op on a missing client, which would otherwise surface much later as an empty overlay.

Parameters:: **data – The backend’s response fields, plus optional client (HTTP client) and pipeline_id.

element(name: str) → ionworks.pipeline.BaseResults[source]¶

Return the typed ionworks.pipeline result for a named element.

Decodes element_results[name] into the matching ionworks.pipeline result class (see _coerce_element_result()) and caches it. The overlay / trace / posterior fetchers are bound to the element’s job.

Parameters:: name (str) – The element name (the key used in element_results).
Returns:: The typed result object.
Return type:: ionworks.pipeline.BaseResults
Raises:: ValueError – If no element with the given name exists in the pipeline.

property results: dict[str, TypeAliasForwardRef('ionworks.pipeline.BaseResults')]¶: Typed results for every element, keyed by element name (cached).

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

model_post_init(context: Any, /) → None¶

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.

Args:: self: The BaseModel instance. context: The context.

class ionworks.api.pipeline.PipelineClient(client: Any)[source]¶

Bases: object

Client for creating and managing pipeline workflows.

__init__(client: Any) → None[source]¶

Initialize the pipeline client.

Parameters:: client (Any) – The HTTP client to use for API requests.

create(config: TypeAliasForwardRef('ionworks.pipeline.Pipeline') | dict[str, Any], *, project_id: str | None = None, name: str | None = None, description: str | None = None, options: dict[str, Any] | None = None) → PipelineSubmissionResponse[source]¶

Run a complete pipeline with the given configuration.

Parameters:

config (ionworks.pipeline.Pipeline or dict[str, Any]) – Pipeline configuration. Either an iw.pipeline.Pipeline schema instance (constructed via iw.pipeline.Pipeline(elements=...)) or a dict with elements, name, description and SDK-only fields such as project_id/options. Dicts are validated against iw.pipeline.Pipeline before submission so shape errors surface locally.
project_id (str, optional) – Project to submit to. Falls back to a project_id field on config (dict form), then to the PROJECT_ID env var.
name (str, optional) – Submission name override. Falls back to the schema’s name field.
description (str, optional) – Submission description override. Falls back to the schema’s description field.
options (dict[str, Any], optional) – Submission options (e.g. {"live_progress_updates": True}). Falls back to config["options"] (dict form).

Returns:

The pipeline submission response.

Return type:

PipelineSubmissionResponse

Raises:

ValueError – If the configuration is invalid.

update(pipeline_id: str, name: str | None = None, description: str | None = None) → PipelineSubmissionResponse[source]¶

Partially update a pipeline’s name and/or description.

This is a metadata-only update: it cannot modify the pipeline’s config or affect a running job.

Parameters:

pipeline_id (str) – The pipeline ID to update.
name (str | None, optional) – New name. Omit (or pass None) to leave unchanged.
description (str | None, optional) – New description. Omit (or pass None) to leave unchanged.

Returns:

The updated record.

Return type:

PipelineSubmissionResponse

Raises:

ValueError – If neither name nor description is provided.

List pipelines for a project.

Parameters:

project_id (str | None, optional) – The project id to filter pipelines. If not provided, uses the project_id set on the Ionworks client or the IONWORKS_PROJECT_ID environment variable.
limit (int | None, optional) – Maximum number of pipelines to return. If not provided, returns all pipelines (up to the API’s default limit).
offset (int | None, optional) – Number of pipelines to skip for pagination. Sent to the API as skip — the /pipelines route’s wire name for this parameter — while keeping offset as the Python name for consistency with the SDK’s other list() methods.
id (str | None, optional) – Optional filters. Pass values directly for exact match or use Supabase operator syntax (e.g. "ilike.%foo%", "in.(completed,failed)").
name (str | None, optional) – Optional filters. Pass values directly for exact match or use Supabase operator syntax (e.g. "ilike.%foo%", "in.(completed,failed)").
description (str | None, optional) – Optional filters. Pass values directly for exact match or use Supabase operator syntax (e.g. "ilike.%foo%", "in.(completed,failed)").
status (str | None, optional) – Optional filters. Pass values directly for exact match or use Supabase operator syntax (e.g. "ilike.%foo%", "in.(completed,failed)").
created_by_email (str | None, optional) – Optional filters. Pass values directly for exact match or use Supabase operator syntax (e.g. "ilike.%foo%", "in.(completed,failed)").
created_at (str | None, optional) – Optional Supabase-operator filter on the timestamp field (e.g. "gte.2025-01-01").
updated_at (str | None, optional) – Optional Supabase-operator filter on the timestamp field (e.g. "gte.2025-01-01").
created_at_gt – str | None, optional Inclusive lower / upper bounds for between-style date queries.
created_at_lt – str | None, optional Inclusive lower / upper bounds for between-style date queries.
updated_at_gt – str | None, optional Inclusive lower / upper bounds for between-style date queries.
updated_at_lt – str | None, optional Inclusive lower / upper bounds for between-style date queries.
order_by (str | None, optional) – Sort column. One of id, name, description, status, created_at, updated_at, created_by_email. Falls back to the API’s default (created_at).
order (str | None, optional) – Sort direction ("asc" or "desc"). Falls back to the API’s default ("desc").

Returns:

A list-like page of pipeline submission responses, with .count and .total for pagination.

Return type:

PaginatedList[PipelineSubmissionResponse]

Raises:

ValueError – If response data is not a list or project_id is missing.

get(job_id: str) → PipelineSubmissionResponse[source]¶

Get the pipeline response for the given job id.

Parameters:: job_id (str) – The job id.
Returns:: The pipeline submission response.
Return type:: PipelineSubmissionResponse

result(job_id: str) → PipelineResponse[source]¶

Get the result for the given job id.

Parameters:: job_id (str) – The job id.
Returns:: The pipeline results.
Return type:: PipelineResponse

list_elements(pipeline_id: str, *, summary: bool = False) → list[dict[str, Any]][source]¶

List a pipeline’s elements in run order, with each one’s config and result.

Pass summary=True for metadata without configuration or result data.

Parameters:: pipeline_id (str) – The pipeline whose elements to list.
Returns:: One record per element: name, element_order, element_type, status, error, job_id, job_config (what it was run with) and result (including the parameter_values it produced).
Return type:: list[dict[str, Any]]

get_element_job_config(element_id: str) → dict[str, Any][source]¶

Get one element’s full job_config.

list_elements() leaves the larger keys (parameters, optimizer, cost) out of each element’s job_config; this returns them: the fixed parameter values, the fitted parameters and their bounds, and the optimizer settings.

Parameters:: element_id (str) – The element’s id, as list_elements() returns it.
Returns:: The element’s job_config, or {} for an element that does not exist or that you cannot see.
Return type:: dict[str, Any]

cancel(pipeline_id: str) → PipelineSubmissionResponse[source]¶

Cancel a running pipeline and all its non-terminal elements.

Parameters:: pipeline_id (str) – The pipeline ID to cancel.
Returns:: The updated record (status will be canceled if cancellation took effect; otherwise the current state is returned).
Return type:: PipelineSubmissionResponse

delete(pipeline_id: str) → None[source]¶

Delete a pipeline, its elements, associated jobs, and storage files.

Parameters:: pipeline_id (str) – The pipeline ID to delete.

get_element_metadata(pipeline_id: str, element_name: str, elements: list[dict[str, Any]] | None = None) → dict[str, Any][source]¶

Fetch the metadata blob for a named element of a pipeline.

Locates the element by name, then delegates to client.job.get_metadata. Use it for fields stripped from element.result and persisted to storage instead — for example the validation_plot_config a validation element writes.

Not for a validation element’s time series: those are stored per objective rather than on this blob. Use get_element_series_channels() and get_element_series(), which cost one objective rather than all of them.

Parameters:

pipeline_id (str) – The pipeline whose element metadata to fetch.
element_name (str) – The name of the element within the pipeline — the key used in the elements dict at submission time. Element names are user-chosen and unique per pipeline (a pipeline may run multiple validation elements under names like "validate_pristine" and "validate_aged").
elements (list[dict], optional) – Pre-fetched elements list from GET /pipelines/{id}/elements. Pass this when pulling metadata for several elements of the same pipeline to avoid re-fetching the list on every call. When omitted, the list is fetched fresh.

Returns:

The parsed metadata payload for the element’s job.

Return type:

dict[str, Any]

Raises:

ValueError – If the pipeline has no element with the given name, or that element has no associated job (e.g. it never ran).

get_element_series_channels(pipeline_id: str, element_name: str, elements: list[dict[str, Any]] | None = None) → dict[str, dict[str, list[str]]][source]¶

List an element’s validation-series objectives and their channels.

Delegates to get_series_channels() for the element’s job. Names only, no samples — call it before get_element_series() to learn which objectives exist, which channels each carries, and which of the two solves it has.

Parameters:

pipeline_id (str) – The pipeline whose element to describe.
element_name (str) – Name of the element within the pipeline.
elements (list[dict], optional) – Pre-fetched elements list, as for get_element_metadata().

Returns:

{objective: {source: [channel, ...]}}.

Return type:

dict[str, dict[str, list[str]]]

get_element_series(pipeline_id: str, element_name: str, objective: str, elements: list[dict[str, Any]] | None = None, **kwargs: Any) → dict[str, list][source]¶

Get one objective’s model-vs-data series from a named element.

Delegates to get_series() for the element’s job. This is the way to read a validation element’s time series: it reads one objective rather than the element’s whole metadata blob.

Parameters:

pipeline_id (str) – The pipeline whose element to read.
element_name (str) – Name of the element within the pipeline.
objective (str) – Objective name, as listed by get_element_series_channels().
elements (list[dict], optional) – Pre-fetched elements list, as for get_element_metadata().
**kwargs – Passed through to get_series() — source, columns, x_column, max_points, x_min, x_max.

Returns:

One list per channel, all of equal length, keyed by channel name.

Return type:

dict[str, list]

wait_for_completion(pipeline_id: str, timeout: int = 600, poll_interval: int = 2, verbose: bool = True, raise_on_failure: bool = True) → PipelineSubmissionResponse[source]¶

Wait for a pipeline to complete by polling until done or timeout.

Parameters:

pipeline_id (str) – The pipeline ID to wait for.
timeout (int, optional) – Maximum time to wait in seconds (default: 600).
poll_interval (int, optional) – Time between polls in seconds (default: 2).
verbose (bool, optional) – Whether to print status updates (default: True).
raise_on_failure (bool, optional) – Whether to raise IonworksError when the pipeline fails or is canceled (default: True).

Returns:

The terminal pipeline response (failed / canceled included when raise_on_failure=False).

Return type:

PipelineSubmissionResponse

Raises:

TimeoutError – If timeout is reached before the pipeline reaches a terminal state.
IonworksError – If the pipeline failed or was canceled and raise_on_failure is True.