Error Handling in Practice

What you’ll learn¶

How the framework catches and records runtime failures
Finding and reading failure logs
Partial failures with FailurePolicy (CONTINUE vs FAIL_FAST)
Empty input cascades from strict filters
Validation errors vs runtime errors

Prerequisites: First Pipeline, Exploring Results
Estimated time: 15 minutes
GPU required: No.

Pipelines encounter errors in different ways: bad input data, strict filters that eliminate everything, invalid configuration. The framework handles each case differently — some errors are recorded and the pipeline continues, others fail fast. This tutorial walks through each scenario so you know what to expect and where to look.

from __future__ import annotations

from artisan.operations.curator import Filter, IngestData
from artisan.operations.examples import DataGeneratorWithMetrics, DataTransformer
from artisan.orchestration import PipelineManager
from artisan.schemas.enums import FailurePolicy
from artisan.utils import tutorial_setup
from artisan.visualization import inspect_pipeline

A step that fails¶

The simplest way to trigger a runtime failure is to feed bad data into an operation that expects numeric values. We’ll create a CSV with "INVALID" in a numeric column, ingest it with IngestData, then pass it to DataTransformer. The ingestion succeeds (it just stores bytes), but the transformer fails when it tries float("INVALID").

env = tutorial_setup("error_handling")

# Create a CSV with a non-numeric value in the 'score' column
bad_csv = env.working_root / "bad_data.csv"
bad_csv.write_text("id,x,y,z,score\n0,1.0,2.0,3.0,INVALID\n")
print(f"Bad CSV: {bad_csv}")
print(bad_csv.read_text())

pipeline = PipelineManager.create(
    name="error_handling",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)
output = pipeline.output

# Step 0: Ingest the bad CSV (succeeds — just stores bytes)
step_0 = pipeline.run(
    operation=IngestData,
    name="ingest",
    inputs=[str(bad_csv)],
)

# Step 1: Transform — fails when parsing the non-numeric value
step_1 = pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("ingest", "data")},
)

result = pipeline.finalize()
print(f"Pipeline complete: success={result['overall_success']}")

print(
    f"Step 0 (ingest):    success={step_0.success}, succeeded={step_0.succeeded_count}, failed={step_0.failed_count}"
)
print(
    f"Step 1 (transform): success={step_1.success}, succeeded={step_1.succeeded_count}, failed={step_1.failed_count}"
)

inspect_pipeline(env.delta_root)

Notice that inspect_pipeline shows "ok" for both steps — it marks any completed step as ok, even with partial failures. To detect failures, check StepResult.failed_count or StepResult.has_failures directly.

Finding and reading failure logs¶

When an execution fails, the framework writes a human-readable log file at <runs_dir>/logs/failures/step_N_opname/run_id.log. Each file contains the run ID, operation name, full traceback, and the last 100 lines of tool output.

logs_dir = env.runs_dir / "logs" / "failures"
if logs_dir.exists():
    for step_dir in sorted(logs_dir.iterdir()):
        for log_file in sorted(step_dir.iterdir()):
            print(f"--- {log_file.relative_to(env.runs_dir)} ---")
            print(log_file.read_text())
else:
    print("No failure logs found")

The log shows the exact error (ValueError: could not convert string to float: 'INVALID') and the traceback pointing to data_transformer.py. The execution run ID in the log matches the execution_run_id column in the executions Delta table, so you can cross-reference failures programmatically.

import polars as pl

from artisan.schemas.enums import TablePath

executions_path = env.delta_root / TablePath.EXECUTIONS
df = pl.read_delta(str(executions_path))
df.select(
    "execution_run_id", "origin_step_number", "operation_name", "success", "error"
)

Partial failures with FailurePolicy¶

Real pipelines process many inputs at once. When some fail and others succeed, the FailurePolicy controls what happens:

Policy	Behavior
`CONTINUE` (default)	Log failures, commit successful results, report counts in StepResult
`FAIL_FAST`	Stop on first failure, raise `RuntimeError`, no commit

We’ll create 5 CSVs where 2 have invalid data, then run with each policy.

env_partial = tutorial_setup("error_handling_partial")

# Create 5 CSVs — indices 1 and 3 have non-numeric values
csv_dir = env_partial.working_root / "csvs"
csv_dir.mkdir(parents=True, exist_ok=True)

csv_paths = []
for i in range(5):
    path = csv_dir / f"data_{i:03d}.csv"
    if i in (1, 3):
        path.write_text(f"id,x,y,z,score\n{i},1.0,2.0,3.0,INVALID\n")
    else:
        path.write_text(f"id,x,y,z,score\n{i},1.0,2.0,3.0,0.5\n")
    csv_paths.append(path)

print(f"Created {len(csv_paths)} CSVs (indices 1, 3 have bad data)")

pipeline = PipelineManager.create(
    name="partial_continue",
    delta_root=env_partial.delta_root,
    staging_root=env_partial.staging_root,
    working_root=env_partial.working_root,
)
output = pipeline.output

step_0 = pipeline.run(
    operation=IngestData,
    name="ingest",
    inputs=[str(p) for p in csv_paths],
)

# Default policy is CONTINUE — failures are logged, successes committed
step_1 = pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("ingest", "data")},
)

result = pipeline.finalize()

print(f"Step 1: succeeded={step_1.succeeded_count}, failed={step_1.failed_count}")
print(f"Pipeline success={result['overall_success']}")

inspect_pipeline(env_partial.delta_root)

With CONTINUE, the step completed with 3 successes and 2 failures. The 3 successful outputs were committed to Delta Lake. Any downstream step would receive only those 3 artifacts.

Now let’s try the same pipeline with FAIL_FAST:

env_fast = tutorial_setup("error_handling_failfast")

pipeline = PipelineManager.create(
    name="partial_failfast",
    delta_root=env_fast.delta_root,
    staging_root=env_fast.staging_root,
    working_root=env_fast.working_root,
    failure_policy=FailurePolicy.FAIL_FAST,
)
output = pipeline.output

step_0 = pipeline.run(
    operation=IngestData,
    name="ingest",
    inputs=[str(p) for p in csv_paths],
)

try:
    step_1 = pipeline.run(
        operation=DataTransformer,
        name="transform",
        inputs={"dataset": output("ingest", "data")},
    )
except RuntimeError as e:
    print(f"FAIL_FAST raised RuntimeError: {e}")

With FAIL_FAST, the pipeline stops on the first failure and raises a RuntimeError. No results are committed. Use FAIL_FAST when partial results are meaningless and you want to catch problems early.

Empty input cascades¶

When a filter eliminates all artifacts, downstream steps have nothing to process. The pipeline doesn’t crash — it skips those steps and records why.

We’ll use a DataGeneratorWithMetrics that produces mean_score values in [0, 1], then filter with an impossible threshold of > 100.

env_cascade = tutorial_setup("error_handling_cascade")

pipeline = PipelineManager.create(
    name="empty_cascade",
    delta_root=env_cascade.delta_root,
    staging_root=env_cascade.staging_root,
    working_root=env_cascade.working_root,
)
output = pipeline.output

step_0 = pipeline.run(
    operation=DataGeneratorWithMetrics,
    name="generate",
    params={"count": 5, "seed": 42},
)

step_1 = pipeline.run(
    operation=Filter,
    name="filter",
    inputs={"passthrough": output("generate", "datasets")},
    params={
        "criteria": [{"metric": "mean_score", "operator": "gt", "value": 100}],
    },
)

step_2 = pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("filter", "passthrough")},
)

result = pipeline.finalize()
print(f"Pipeline complete: success={result['overall_success']}")

inspect_pipeline(env_cascade.delta_root)

for step in [step_0, step_1, step_2]:
    skipped = step.metadata.get("skipped", False)
    if skipped:
        reason = step.metadata["skip_reason"]
        print(f"Step {step.step_number} ({step.step_name}): skipped — {reason}")
    else:
        print(
            f"Step {step.step_number} ({step.step_name}): {step.succeeded_count} succeeded"
        )

The filter passed zero artifacts, so step 2 was skipped with skip_reason="empty_inputs". If there were more downstream steps, they would show "pipeline_stopped".

Validation errors¶

Not all errors happen during execution. If you pass invalid parameters, Pydantic catches the problem at pipeline.run() time — before any worker is dispatched. This is intentional: configuration mistakes fail fast before wasting compute.

import pydantic

env_val = tutorial_setup("error_handling_validation")

pipeline = PipelineManager.create(
    name="validation_demo",
    delta_root=env_val.delta_root,
    staging_root=env_val.staging_root,
    working_root=env_val.working_root,
)

try:
    pipeline.run(
        operation=DataTransformer,
        name="transform",
        inputs={"dataset": []},
        params={"scale_factor": "not_a_number"},
    )
except (pydantic.ValidationError, Exception) as e:
    print(f"{type(e).__name__}: {e}")

Validation errors are raised immediately — no failure log is written, no execution is recorded in Delta Lake. The error message tells you exactly which parameter was wrong and what was expected.

This contrasts with runtime errors (like the bad CSV above), which happen inside workers, are recorded in the executions table, and produce failure log files.

Summary¶

Error type	When caught	Where recorded	Failure log?
Validation	At `pipeline.run()`	Not recorded — exception raised	No
Runtime (execute)	In worker	Executions table + failure log	Yes
Empty inputs	At step start	Steps table (skipped)	No

Key points:

inspect_pipeline shows "ok" for completed steps even with partial failures — use StepResult.failed_count for the full picture
Failure logs live at <runs_dir>/logs/failures/step_N_op/run_id.log
FailurePolicy.CONTINUE (default) commits successful results and records failures; FAIL_FAST raises on the first failure
Empty filter results cascade as skipped steps, not failures
Validation errors fail fast before execution — no log, no record

Next steps¶

Storage Layout and Logging — Directory layout, Delta Lake tables, and logging configuration
Error Handling — Design rationale for failure containment
Resume and Caching — How caching interacts with failures during re-runs