Python API

Submit, monitor, list, and retrieve results for pipeline jobs with the ionworks Python client.

The ionworks Python package exposes client.pipeline for running and managing pipelines. For installation and authentication, see the Python API client page.

Submitting a pipeline

import ionworks as iw

# Reads IONWORKS_API_KEY and IONWORKS_PROJECT_ID from the environment.
client = iw.api.Client()

pipeline = iw.pipeline.Pipeline(
    {
        "known": iw.pipeline.direct_entries.DirectEntry(
            parameters={"Ambient temperature [K]": 298.15},
        ),
        "Q_pos": iw.pipeline.calculations.ElectrodeCapacity(electrode="positive"),
    },
    name="Capacity pipeline",
)

submission = client.pipeline.create(pipeline)
print(f"Pipeline ID: {submission.id}")
print(f"Status: {submission.status}")

client.pipeline.create() accepts either an iw.pipeline.Pipeline schema instance or the dict returned by .to_config(). Schema instances are validated locally before submission, so shape errors surface immediately.

Serializing a pipeline to JSON

Use .to_config() when you want to inspect, cache, or transport the pipeline payload as JSON — for example to review it before submission, commit it to version control, or hand it off to another process.

import json

config = pipeline.to_config()

# Inspect or save
with open("pipeline_config.json", "w") as f:
    json.dump(config, f, indent=2)

# Submit later, or from another process
submission = client.pipeline.create(config)

.to_config() is the only supported serializer. It emits the discriminators pipeline elements and objectives need (top-level elements are keyed on element_type, and nested schemas carry their own type), and emits each field under its wire name (for example, data_input → data).

Do not use Pydantic’s model_dump() to build API payloads. model_dump() drops these discriminators and emits Python attribute names instead of wire names, so it produces a dict the API may reject.

Overriding submission metadata

create() accepts optional project_id, name, description, and options kwargs that override any values carried on the schema:

submission = client.pipeline.create(
    pipeline,
    project_id="your-project-id",
    name="Custom run name",
    options={"live_progress_updates": True},
)

When project_id is omitted, the client falls back to the default configured on iw.api.Client(...) or the IONWORKS_PROJECT_ID environment variable.

Data references in pipelines

Use these prefixes to reference data sources in pipeline configs:

Prefix	Example	Description
db:	"db:measurement-id"	Reference an uploaded measurement by ID
file:	"file:data.csv"	Load a local CSV file
folder:	"folder:data_dir/"	Load from a local directory

For inline DataFrames in pipeline configs, there is a 1,000-row limit. Upload larger datasets as measurements first, then reference them with db:measurement-id.

A db: measurement that is still being recorded grows as rows are appended, so an unpinned config reads more data on a later run than it did on the first. Add a time_range alongside data to pin the run to a fixed window of elapsed seconds:

{"data": "db:measurement-id", "time_range": {"start": 0, "end": 3600}}

Give duration instead of end to fix the window’s length from start: {"start": 0, "duration": 86400} is the first day of the test, however long it eventually runs. iw.pipeline.TimeRangeSpec(start=0, duration=86400) builds the same window. It is stored as the equivalent end, and passing both end and duration is rejected.

See pinning a measurement to a time window.

The folder: scheme expects a directory containing time_series and steps files. Both .parquet and .csv are supported, and parquet is preferred when both are present. For example, a folder with time_series.parquet and steps.parquet (or .csv) loads correctly.

PyBaMM model support

You can hand an objective a PyBaMM model object directly instead of naming a built-in model. It is serialized for you when the config is built — there is no registration step and no string name to look up.

import pybamm
import ionworks as iw

objective = iw.pipeline.objectives.CurrentDriven(
    data_input="file:path/to/discharge.csv",
    options={"model": pybamm.lithium_ion.DFN()},
)

Model options travel with the object, so a configured model arrives configured:

import ionworks as iw

model = pybamm.lithium_ion.DFN(options={"thermal": "lumped"})

objective = iw.pipeline.objectives.CurrentDriven(
    data_input="file:path/to/discharge.csv",
    options={"model": model},
)

Waiting for completion

submission = client.pipeline.wait_for_completion(
    submission.id,
    timeout=600,        # seconds (default: 600)
    poll_interval=2,    # seconds between polls (default: 2)
    verbose=True,       # print status updates (default: True)
)

Returns once the pipeline reaches a terminal status — completed, failed, or canceled. A cancelled pipeline is terminal, so it returns immediately rather than polling to the timeout.

Pass raise_on_failure=False to get the terminal submission response back instead of raising when the pipeline fails or is cancelled.

Otherwise a failed or cancelled pipeline raises an IonworksError whose error_code is "JOB_FAILED"; its data["detail"] carries the failure’s job_error_code and job_error to branch on; see Failed jobs.

Retrieving results

Each element returns a typed result object. Ask the pipeline for an element by the name you gave it, index the fitted parameters, and plot:

result = client.pipeline.result(submission.id)
fit = result.element("fit")

print(fit.parameter_values["Positive particle diffusivity [m2.s-1]"])

fit.plot_fit_results()   # measured vs fitted, one figure per objective
fit.plot_trace()         # optimizer cost + parameter convergence

The class you get back reflects what the element did — OptimizationResult, PosteriorResult, ValidationResult and so on, all sharing a BaseResults base — so parameter_values, to_config() and the plots work the same way on any of them. result.results returns every element at once, keyed by name.

The overlay and the optimizer trace are fetched on first access, not when the result is built, so reading parameter_values costs nothing extra.

The raw payloads are still there if you want them — result.result for the pipeline’s final parameter values and result.element_results for per-element dicts, keyed exactly as you passed them to iw.pipeline.Pipeline(elements=...).

Element metadata

Elements write large fields (a validation’s full per-point comparison, for example) to a metadata blob rather than into element_results. The result object already reads what it needs from there; fetch the blob yourself only for fields it does not expose:

metadata = client.pipeline.get_element_metadata(submission.id, "validate")

Element configs

list_elements returns a pipeline’s elements in run order, each with what it was run with (job_config) and what it produced (result, including its parameter_values):

for el in client.pipeline.list_elements(submission.id):
    print(el["element_order"], el["name"], el["element_type"], el["status"])

# Metadata only, without configs or results
client.pipeline.list_elements(submission.id, summary=True)

Each element’s job_config leaves out parameters, optimizer and cost. get_element_job_config returns them for one element, by its id: the fixed parameter values, the fitted parameters and their bounds, and the optimizer settings.

config = client.pipeline.get_element_job_config(el["id"])

It returns {} for an element that does not exist or that you cannot see.

Data-fit parameter trace

fit.plot_trace() draws the optimizer’s per-iteration progress, and fit.trace gives the raw records if you would rather plot them yourself — see Inspecting the parameter trace for the schema. Neither needs the element’s job ID; the result object resolves that for you.

Data-fit model-vs-data plot data

fit.plot_fit_results() above covers the common case. Drop to client.job.get_plot_data when you want the raw traces — to feed your own plotting stack, or to refetch detail as a user zooms. It returns the model-vs-data traces for a data-fit job — the same overlay Studio renders on the fit’s results page. A data-fit re-runs a validation on its best-fit parameters and stores the overlay in its own metadata, so you can fetch it directly from the fit’s job ID without adding a separate Validation element or parsing the raw metadata blob.

The endpoint is keyed by job ID, which the elements listing carries:

elements = client.get(f"/pipelines/{submission.id}/elements")
fit_job_id = next(e["job_id"] for e in elements if e["name"] == "fit")

plot = client.job.get_plot_data(
    fit_job_id,
    objective_name="1C",   # key used in the DataFit's objectives mapping
)

Traces are decimated server-side to at most max_points points per series (default 2000, range 100–80000). For semantic zoom — refetching more detail as a user zooms in — pass x_min and x_max set to the current viewport:

zoomed = client.job.get_plot_data(
    fit_job_id,
    objective_name="1C",
    x_min=1200.0,
    x_max=1800.0,
    max_points=5000,
)

Parameter	Description
job_id	The data-fit job whose overlay to fetch.
objective_name	Key in the DataFit’s objectives mapping (for example, "1C" in the example above).
plot_index	Index into the objective’s list of plots when it defines several. Defaults to 0.
max_points	Maximum points returned per trace (100–80000). Defaults to 2000.
x_min, x_max	Inclusive x-range bounds. Omit for the full range.

Validation series

get_plot_data above reads a data fit’s stored overlay. For a validation job — or to read one objective of a fit without pulling the whole job’s metadata — use client.job.get_series. Reading one objective costs what that objective costs, rather than what the whole fit costs, so it stays usable on large runs where get_metadata does not.

Ask what the job recorded first. get_series_channels returns names only, no samples:

channels = client.job.get_series_channels(job_id)
# {"1C": {"optimal": ["Time [s]", "Voltage [V]"], "baseline": [...]}}

Each objective reports which of the two solves it carries: "optimal" (the fitted parameters) and, when the job ran one, "baseline". Only genuine per-sample channels are listed, so a scalar recorded alongside the series does not appear.

Then read one objective:

series = client.job.get_series(
    job_id,
    objective="1C",          # a key of the mapping above
    source="optimal",        # or "baseline"
    x_min=1200.0,
    x_max=1800.0,
    max_points=5000,
)
# {"Time [s]": [...], "Voltage [V]": [...]}

The window is applied before the row budget, so the points you get are spent on the range you asked for rather than thinned across the whole run.

Parameter	Description
job_id	The data-fit, validation, or optimization job to read.
objective	Objective name — a key of the mapping get_series_channels returns.
source	Which solve to read: "optimal" (default) or "baseline".
columns	Channels to return besides x_column. Defaults to every channel the objective carries. Sent as one comma-separated parameter, so a channel whose name contains a comma cannot be requested.
x_column	Channel plotted on x, and the one windowed and binned on. Defaults to the objective’s own x — "Time [s]" for a time-domain fit, "Frequency [Hz]" for an impedance one — so it rarely needs setting.
max_points	Row ceiling for the response. Omit to return the window whole, which is the only way to be certain nothing was decimated.
x_min, x_max	Window bounds, in the units of x_column. Omit for the full range.

An objective, source, x_column, or column the job does not carry raises IonworksError — which is why it is worth calling get_series_channels first rather than guessing a name.

Listing pipelines

import ionworks as iw

# Defaults to the project on iw.api.Client(...) or IONWORKS_PROJECT_ID
pipelines = client.pipeline.list()

# Override the project for a single call
pipelines = client.pipeline.list(project_id="other-project-id")

# Limit the number of results
pipelines = client.pipeline.list(limit=10)

# Filter, order, and paginate server-side
active = client.pipeline.list(status="in.(pending,running)")
matches = client.pipeline.list(name="ilike.%diffevo%")
recent = client.pipeline.list(order_by="created_at", order="desc", offset=25)

list() accepts id, name, description, status, created_by_email, created_at and updated_at (plus their _gt / _lt range variants), order_by, order, limit, and offset.

The result is a PaginatedList: it behaves like a list — iterate it, index it, call len() on it — and also carries .items, .count, and .total, where .total counts every matching pipeline rather than the rows in this page.

Getting a single submission

submission = client.pipeline.get(submission.id)
print(submission.status)  # "pending", "running", "completed", "failed", or "canceled"
print(submission.project_id)

End-to-end example

import pybamm
import ionworks as iw

client = iw.api.Client()

pipeline = iw.pipeline.Pipeline(
    {
        "known": iw.pipeline.direct_entries.DirectEntry(
            parameters={"Ambient temperature [K]": 298.15},
        ),
        "fit": iw.pipeline.DataFit(
            objectives={
                "1C": iw.pipeline.objectives.CurrentDriven(
                    data_input="file:examples/api/data/chen_synthetic_1C/time_series.csv",
                    options={"model": pybamm.lithium_ion.SPMe()},
                ),
            },
            parameters={
                "Negative particle diffusivity [m2.s-1]": iw.pipeline.Parameter(
                    "Negative particle diffusivity [m2.s-1]",
                    initial_value=2e-14,
                    bounds=(1e-14, 1e-13),
                ),
            },
            cost=iw.pipeline.costs.RMSE(),
            optimizer=iw.pipeline.optimizers.DifferentialEvolution(),
        ),
    },
    name="SPMe diffusivity fit",
)

submission = client.pipeline.create(pipeline)
client.pipeline.wait_for_completion(submission.id)

result = client.pipeline.result(submission.id)
print(result.element("fit").parameter_values)

For more end-to-end examples (entry-only, calculation-only, datafit, validation), see packages/ionworks/examples/api/pipeline/ in the SDK repo.

Migrating from SimplePipeline

SimplePipeline has been removed. Calls to client.simple_pipeline or the /simple_pipelines REST endpoints now fail: the API answers every one with 410 Gone, and past SimplePipeline runs can no longer be fetched. Releases of ionworks made after the removal no longer include iw.pipeline.SimplePipeline or client.simple_pipeline; accessing either raises an AttributeError that links to this section. Use Pipeline instead, mapping each call as shown below.

iw.pipeline.Pipeline takes the same elements that iw.pipeline.SimplePipeline did, without its one-expensive-element limit, and each former client.simple_pipeline method has a client.pipeline method of the same name. Most code ports by swapping the class and the client. The differences are in what comes back and in a few arguments, listed below. A Pipeline run also appears in Studio, and its data fit stores its own model-vs-data overlay.

SimplePipeline (removed)	Pipeline
iw.pipeline.SimplePipeline(elements=..., name=...)	iw.pipeline.Pipeline(elements=..., name=...). Same elements, with no limit on expensive elements.
client.simple_pipeline.create(config, project_id, ...)	client.pipeline.create(config, project_id=..., name=..., options=...). Everything after config is keyword-only. A dict config is checked locally and raises ValueError if invalid, before anything is sent.
client.simple_pipeline.wait_for_completion(id, timeout=...)	client.pipeline.wait_for_completion(id, timeout=...). Returns the status only; fetch results separately (next row).
result.result["parameter_values"], result.result_object	client.pipeline.result(id).element("<element name>") returns a typed result with .parameter_values and .plot_fit_results(). A ValidationResult also has .summary_stats.
result.result["cost"]	fit.cost on that same typed result: the lowest cost any start reached. None for an ArrayDataFit, and for a Pipeline run that finished before mid-September 2026.
.job_id, .created_at, .updated_at, .created_by_email on a get / list item	Not on a client.pipeline item, which carries id, name, description, status, error, is_terminal and is_failed.
client.simple_pipeline.list(...), 10 per page by default	client.pipeline.list(...), up to 100 per page by default. Pass limit=10 to keep the old page size. Same filters and ordering.
client.simple_pipeline.get / update / cancel / delete	client.pipeline.get / update / cancel / delete
simple_pipeline.completed / .failed / .canceled webhook events	pipeline.completed / .failed / .canceled. The old events were removed from existing subscriptions, and a webhook subscribed to nothing else was disabled. Choose pipeline events for it, then re-enable it.
POST /simple_pipelines, with the elements under config	POST /pipelines, with elements at the top level of the body beside project_id and name. Read results from GET /pipelines/{id}/result, under element_results["<element name>"]. Every other /simple_pipelines route had the same path under /pipelines.
source_simple_pipeline_id on analyses and material property datasets	Use source_pipeline_id for new records. The old field is no longer returned and is ignored if sent. Existing values were not moved to source_pipeline_id: a record whose only source was a SimplePipeline has source_label set to Simple pipeline "<name>" (<id>) instead, unless it already had a label. If the run was in another project or had been deleted, the label is Simple pipeline <id>.
GET /simple_pipelines?project_id=...&limit=10&offset=20	GET /pipelines?project_id=...&limit=10&skip=20. Rename the REST pagination parameter offset to skip; leaving offset repeats the first page because it is ignored.
REST list response: items, count, total	REST list response: pipelines, total. Read the page from pipelines and use its length for the page count; count is no longer returned.

The SDK handles these REST listing differences: client.pipeline.list(offset=...) still returns a PaginatedList with items, count, and total.

The elements are identical before and after. The class, the client, and how you read the result change:

import ionworks as iw

client = iw.api.Client()

elements = {
    "fit": iw.pipeline.DataFit(
        objectives={
            "cycle": iw.pipeline.objectives.CurrentDriven(
                data_input="file:path/to/discharge.csv", options={"model": "SPM"}
            )
        },
        parameters={
            "Negative particle diffusivity [m2.s-1]": iw.pipeline.Parameter(
                "Negative particle diffusivity [m2.s-1]",
                bounds=(1e-14, 1e-13),
                initial_value=2e-14,
            )
        },
        cost=iw.pipeline.costs.RMSE(),
    ),
}

Before (removed)

pipeline = iw.pipeline.SimplePipeline(elements=elements, name="My Fit")

sp = client.simple_pipeline.create(pipeline, name=pipeline.name)
result = client.simple_pipeline.wait_for_completion(sp.id, timeout=600)

print(result.result["parameter_values"])

After

pipeline = iw.pipeline.Pipeline(elements=elements, name="My Fit")

submission = client.pipeline.create(pipeline, name=pipeline.name)
client.pipeline.wait_for_completion(submission.id, timeout=600)

fit = client.pipeline.result(submission.id).element("fit")
print(fit.parameter_values)
fit.plot_fit_results()

For a validation, ask for the Validation element by name the same way and read .summary_stats from the ValidationResult it returns. See Retrieving results for everything else a typed result exposes.