Pipeline stage

Skill zakelfassi/skills-driven-development/examples/data-pipeline/skills/pipeline-stage

Agents that learn by doing — and remember how they did it. A methodology for AI agents to create, evolve, and share reusable skills.

Install

npx -y skills add zakelfassi/skills-driven-development --skill pipeline-stage

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

One thing to look at

17 stars17 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

What its author says it does

Copied from the file, not written here

Scaffold a new transform stage in the data pipeline — create the transformation script, schema contract, idempotency logic, and tests. Use when adding a new dbt model or pandas transform, when a new business metric needs a dedicated stage, or when asked to "add a {name} stage to the pipeline".

SKILL.md

3.8 KB, as published. Nobody here has run it

Pipeline Stage

Create a new transform stage with idempotency guarantees, a schema contract, and tests.

Inputs

Stage name (snake_case, e.g., customer_ltv)
Input tables/models (list of upstream stage names or raw tables)
Output table name (usually matches stage name)
Grain (the primary key or unique key, e.g., customer_id, (order_id, date))
Layer (staging, intermediate, marts)

Steps

Create the transform file

For dbt:

models/{layer}/{stage_name}.sql

{{ config(
    materialized='table',
    unique_key='{grain}'
) }}

select
    {grain},
    -- TODO: add business logic
    current_timestamp as updated_at
from {{ ref('{input_table}') }}

For pandas ETL:

pipelines/transforms/{stage_name}/transform.py

def run(df: pd.DataFrame) -> pd.DataFrame:
    """Transform {input_table} → {stage_name}."""
    # TODO: add business logic
    return df

Define the schema contract Create models/{layer}/schema/{stage_name}.yaml (dbt) or pipelines/transforms/{stage_name}/schema.py:
```
- name: {stage_name}
  columns:
    - name: {grain}
      tests:
        - unique
        - not_null
```
Every non-nullable column must have not_null test; every unique key must have unique test.
Add idempotency logic
- For materialized='table': dbt handles full replacement — no extra work.
- For incremental models: use is_incremental() filter on updated_at or an event timestamp.
- For pandas: the output must be deterministic given the same input; add a dedup step on {grain}.
Write tests
```
tests/transforms/test_{stage_name}.py
```
Required tests:
- Input fixture → expected output shape (column names, row count)
- Idempotency: running twice produces identical output
- Null check: no nulls in required columns after transform

Register in the pipeline DAG Add the stage after its upstream dependencies:

# dags/pipeline.py
{stage_name}_task = DbtRunOperator(
    task_id="{stage_name}",
    models="{stage_name}",
)
{upstream_task} >> {stage_name}_task

Run locally

dbt run --select {stage_name}
dbt test --select {stage_name}
# or for pandas:
python -m pytest tests/transforms/test_{stage_name}.py -v

Conventions

Layer hierarchy: raw → staging → intermediate → marts
Never skip a layer (e.g., don't read from raw in a marts model)
All stages have at least one unique + not_null test on the grain column
Incremental models use updated_at as the watermark; add it to every model

Edge Cases

Fan-out (multiple downstream consumers): Create the stage at the intermediate layer; let downstream marts models reference it.
Slowly changing dimension (SCD): Use dbt's snapshot materialization or add valid_from/valid_to columns manually.
Cross-database join: Materialize both inputs to the same database first, then join; cross-database SQL is not portable.
Very wide table (>200 columns): Split into a core model plus an extension model; document the split in the schema YAML.