Writing Your First Operation

What you’ll learn¶

How operations fit into a pipeline (the three-phase lifecycle)
How to define a custom operation from scratch
How to declare inputs, outputs, parameters, and lineage
How to wire your operation into a pipeline and verify the results

Prerequisites: Your First Pipeline. Estimated time: 20 minutes GPU required: No.

What is an operation?¶

An operation is a single unit of computation in a pipeline. In the first tutorial you used built-in operations like DataGenerator and DataTransformer. Now you’ll build your own.

Every operation follows a three-phase lifecycle. The framework calls these methods in order:

  ┌──────────────┐     ┌───────────┐     ┌───────────────┐
  │  preprocess   │ --> │  execute   │ --> │  postprocess   │
  └──────────────┘     └───────────┘     └───────────────┘
   Artifacts → paths    Paths → files     Files → artifacts

Phase	What you do	What the framework provides
preprocess	Extract file paths from input artifacts	`PreprocessInput` with hydrated artifact objects
execute	Run your computation, write output files	`ExecuteInput` with an output directory and the dict from preprocess
postprocess	Wrap output files as draft artifacts	`PostprocessInput` with all files from the output directory

The key idea: preprocess and postprocess are your adapter layers between the framework’s artifact world and execute’s plain-file world. Your core logic lives in execute, which knows nothing about artifacts — it reads files and writes files.

What we’ll build¶

A TextUppercaser operation that:

Takes text files as input
Converts their contents to uppercase
Optionally prepends a header line
Returns the results as new data artifacts with provenance tracked

Simple enough to focus on the mechanics, realistic enough to show the full pattern.

Imports¶

These are the building blocks for any operation.

from __future__ import annotations

from enum import StrEnum
from pathlib import Path
from typing import Any, ClassVar

from pydantic import BaseModel, Field

from artisan.operations.base import OperationDefinition
from artisan.schemas import ArtifactResult
from artisan.schemas.artifact.data import DataArtifact
from artisan.schemas.specs.input_models import (
    ExecuteInput,
    PostprocessInput,
    PreprocessInput,
)
from artisan.schemas.specs.input_spec import InputSpec
from artisan.schemas.specs.output_spec import OutputSpec

Define the operation skeleton¶

Every operation needs four things declared at the class level:

Metadata — name and description (how the framework identifies your operation)
Role enums — InputRole and OutputRole (named slots for data flowing in and out)
Specs — inputs and outputs dicts mapping roles to InputSpec/OutputSpec
Params — a nested Pydantic model for algorithm-specific configuration (optional)

Let’s define all four, then implement the lifecycle methods one at a time.

class TextUppercaser(OperationDefinition):
    """Uppercase the contents of text files."""

    # 1. Metadata
    name = "text_uppercaser"
    description = "Convert text file contents to uppercase"

    # 2. Role enums — must match the keys in inputs/outputs
    class InputRole(StrEnum):
        DOCUMENT = "document"

    class OutputRole(StrEnum):
        DOCUMENT = "document"

    # 3. Specs
    inputs: ClassVar[dict[str, InputSpec]] = {
        InputRole.DOCUMENT: InputSpec(artifact_type="data", required=True),
    }
    outputs: ClassVar[dict[str, OutputSpec]] = {
        OutputRole.DOCUMENT: OutputSpec(
            artifact_type="data",
            infer_lineage_from={"inputs": ["document"]},
        ),
    }

    # 4. Parameters (optional)
    class Params(BaseModel):
        add_header: bool = Field(default=False, description="Prepend a header line")

    params: Params = Params()

    # -- Lifecycle methods (next cells) --

    def preprocess(self, inputs: PreprocessInput) -> dict[str, Any]: ...

    def execute(self, inputs: ExecuteInput) -> Any: ...

    def postprocess(self, inputs: PostprocessInput) -> ArtifactResult: ...


print(f"Registered: {TextUppercaser.name}")
print(f"Input roles:  {list(TextUppercaser.inputs.keys())}")
print(f"Output roles: {list(TextUppercaser.outputs.keys())}")

A few things to notice:

InputRole and OutputRole are StrEnums whose values must exactly match the keys in inputs and outputs. The framework validates this at class creation time.
infer_lineage_from={"inputs": ["document"]} tells the framework that each output artifact descends from an input in the "document" role. This is how provenance edges are created automatically — you declare the relationship, the framework records it.
Params is a Pydantic model, so you get validation and serialization for free. Pipeline users pass params as a dict; the framework constructs the model.

Implement preprocess¶

preprocess receives a PreprocessInput containing hydrated artifact objects. Your job: extract what execute needs (usually file paths) and return a plain dict.

This is the most common preprocess pattern — turn artifacts into paths:

def preprocess(self, inputs: PreprocessInput) -> dict[str, Any]:
    return {
        role: [artifact.materialized_path for artifact in artifacts]
        for role, artifacts in inputs.input_artifacts.items()
    }

The materialized_path is a real file path on disk — the framework writes the artifact’s content to a temporary directory before calling preprocess, so your execute method can read the files with standard Python I/O.

Implement execute¶

execute is the core computation — the “black box” that the framework knows nothing about. It receives:

inputs.inputs — the dict returned by preprocess
inputs.execute_dir — a directory where you write output files

Write your results to execute_dir. The framework collects all files in that directory and passes them to postprocess.

def execute(self, inputs: ExecuteInput) -> Any:
    output_dir = inputs.execute_dir
    output_dir.mkdir(parents=True, exist_ok=True)

    for input_path in inputs.inputs["document"]:
        path = Path(input_path)
        text = path.read_text()
        result = text.upper()
        if self.params.add_header:
            result = f"=== UPPERCASED ===\n{result}"
        (output_dir / path.name).write_text(result)

    return {"status": "ok"}

Notice that execute accesses self.params.add_header — parameters are available on the instance. The return value (here {"status": "ok"}) is optional; postprocess can access it via inputs.memory_outputs, but for file-based operations the output files are usually sufficient.

Implement postprocess¶

postprocess converts the files execute wrote into draft artifacts that the framework can commit to Delta Lake. It receives:

inputs.file_outputs — list of all files in execute_dir
inputs.step_number — needed when creating draft artifacts
inputs.memory_outputs — whatever execute returned

def postprocess(self, inputs: PostprocessInput) -> ArtifactResult:
    drafts = [
        DataArtifact.draft(
            content=f.read_bytes(),
            original_name=f.name,
            step_number=inputs.step_number,
        )
        for f in inputs.file_outputs
    ]
    return ArtifactResult(success=True, artifacts={"document": drafts})

DataArtifact.draft() creates a draft artifact — it has no artifact_id yet. The framework finalizes it during commit by computing a content-addressed ID from the bytes.

The dict key "document" in ArtifactResult.artifacts must match your OutputRole value.

Assemble the complete operation¶

Now let’s put all three lifecycle methods together into the real class. This is the complete, working operation.

class TextUppercaser(OperationDefinition):
    """Uppercase the contents of text files."""

    name = "text_uppercaser"
    description = "Convert text file contents to uppercase"

    class InputRole(StrEnum):
        DOCUMENT = "document"

    class OutputRole(StrEnum):
        DOCUMENT = "document"

    inputs: ClassVar[dict[str, InputSpec]] = {
        InputRole.DOCUMENT: InputSpec(artifact_type="data", required=True),
    }
    outputs: ClassVar[dict[str, OutputSpec]] = {
        OutputRole.DOCUMENT: OutputSpec(
            artifact_type="data",
            infer_lineage_from={"inputs": ["document"]},
        ),
    }

    class Params(BaseModel):
        add_header: bool = Field(default=False, description="Prepend a header line")

    params: Params = Params()

    def preprocess(self, inputs: PreprocessInput) -> dict[str, Any]:
        return {
            role: [artifact.materialized_path for artifact in artifacts]
            for role, artifacts in inputs.input_artifacts.items()
        }

    def execute(self, inputs: ExecuteInput) -> Any:
        output_dir = inputs.execute_dir
        output_dir.mkdir(parents=True, exist_ok=True)

        for input_path in inputs.inputs["document"]:
            path = Path(input_path)
            text = path.read_text()
            result = text.upper()
            if self.params.add_header:
                result = f"=== UPPERCASED ===\n{result}"
            (output_dir / path.name).write_text(result)

        return {"status": "ok"}

    def postprocess(self, inputs: PostprocessInput) -> ArtifactResult:
        drafts = [
            DataArtifact.draft(
                content=f.read_bytes(),
                original_name=f.name,
                step_number=inputs.step_number,
            )
            for f in inputs.file_outputs
        ]
        return ArtifactResult(success=True, artifacts={"document": drafts})


print(f"Operation '{TextUppercaser.name}' ready")
print(f"  Inputs:  {list(TextUppercaser.inputs.keys())}")
print(f"  Outputs: {list(TextUppercaser.outputs.keys())}")
print(f"  Params:  {TextUppercaser.Params.model_fields.keys()}")

Wire it into a pipeline¶

Use IngestData to bring text files into the provenance graph, then pass them to our custom operation.

Step 0: IngestData          Step 1: TextUppercaser
┌──────────────────┐       ┌──────────────────────┐
│ Text files on disk │ --->  │ Uppercase + header    │
└──────────────────┘       └──────────────────────┘
   outputs: "data"            inputs: "document"
                              outputs: "document"

from artisan.operations.curator import IngestData
from artisan.orchestration import PipelineManager
from artisan.utils import find_project_root, tutorial_setup

env = tutorial_setup("writing_an_operation")

# Find some CSV files to use as input
PROJECT_ROOT = find_project_root()
SOURCE_FILES = sorted((PROJECT_ROOT / "tests" / "fixtures" / "csv").glob("*.csv"))[:2]

print(f"Source files: {[f.name for f in SOURCE_FILES]}")

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

# Step 0: Ingest text files into the provenance graph
step0 = pipeline.run(
    operation=IngestData, name="ingest", inputs=[str(f) for f in SOURCE_FILES]
)
print(f"Step 0 (IngestData): {step0.succeeded_count} files ingested")

# Step 1: Uppercase them with our custom operation
step1 = pipeline.run(
    operation=TextUppercaser,
    name="uppercase",
    inputs={"document": output("ingest", "data")},
    params={"add_header": True},
)
print(
    f"Step 1 (TextUppercaser): success={step1.success}, produced={step1.succeeded_count} artifacts"
)

Notice the wiring pattern:

output("ingest", "data") creates an OutputReference pointing to the "ingest" step’s "data" role
inputs={"document": output("ingest", "data")} connects that reference to our operation’s "document" input role
The framework resolves the reference at execution time, materializes the artifacts, and calls our lifecycle methods

Inspect the results¶

Let’s verify that both steps ran and our custom operation produced the expected artifacts.

from artisan.visualization import inspect_pipeline

inspect_pipeline(env.delta_root)

The table should show two steps: ingest (step 0) and uppercase (step 1), both with status ok. The produced column confirms that our operation created data artifacts.

from artisan.visualization import inspect_data

# Compare input data (step 0) with output data (step 1)
print("=== Input data (step 0) ===")
display(inspect_data(env.delta_root, step_number=0))

print("\n=== Output data (step 1 — uppercased) ===")
display(inspect_data(env.delta_root, step_number=1))

Provenance¶

The macro graph shows data flow from IngestData to TextUppercaser. Because we declared infer_lineage_from={"inputs": ["document"]}, each output artifact has a provenance edge back to its input — no extra code required.

from artisan.visualization import build_macro_graph

build_macro_graph(env.delta_root)

summary = pipeline.finalize()
print(f"Pipeline '{summary['pipeline_name']}' complete")
print(f"  Steps: {summary['total_steps']}")
print(f"  Success: {summary['overall_success']}")

Recap: the operation anatomy¶

Here’s a cheat sheet for writing operations:

class MyOperation(OperationDefinition):
    # 1. Metadata
    name = "my_operation"
    description = "What this operation does"

    # 2. Roles — StrEnum values must match spec keys
    class InputRole(StrEnum):  ...
    class OutputRole(StrEnum): ...

    # 3. Specs — declare types and lineage
    inputs: ClassVar[dict[str, InputSpec]] = { ... }
    outputs: ClassVar[dict[str, OutputSpec]] = { ... }

    # 4. Params (optional) — algorithm-specific config
    class Params(BaseModel): ...
    params: Params = Params()

    # 5. Lifecycle
    def preprocess(self, inputs)  -> dict:           # artifacts → paths
    def execute(self, inputs)     -> Any:            # paths → files
    def postprocess(self, inputs) -> ArtifactResult: # files → artifacts

The framework validates your class at definition time — if roles don’t match spec keys, or if infer_lineage_from is missing, you get an error before the pipeline runs.

Summary¶

You built a custom operation from scratch:

Declared metadata, roles, specs, and params
Implemented the three-phase lifecycle (preprocess / execute / postprocess)
Set lineage with infer_lineage_from so provenance is tracked automatically
Wired the operation into a pipeline with output("step_name", "role")
Verified the results with inspect_pipeline, inspect_data, and provenance graphs

Next steps¶

Operations Model — Deeper understanding of how operations work and why they’re designed this way
Writing Creator Operations — Task-oriented recipes for advanced patterns (generative ops, multi-input, external tools)
Writing Curator Operations — Filter, Merge, and Ingest patterns
Glossary — Key terms and definitions
Pipeline Patterns — Reusable pipeline topologies