# Pipeline Stage

> 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".

- **Type:** Skill
- **Install:** `agentstack add skill-zakelfassi-skills-driven-development-pipeline-stage`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [zakelfassi](https://agentstack.voostack.com/s/zakelfassi)
- **Installs:** 0
- **Category:** [Data & Analytics](https://agentstack.voostack.com/c/data-and-analytics)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [zakelfassi](https://github.com/zakelfassi)
- **Source:** https://github.com/zakelfassi/skills-driven-development/tree/main/examples/data-pipeline/skills/pipeline-stage
- **Website:** https://zakelfassi.github.io/skills-driven-development/

## Install

```sh
agentstack add skill-zakelfassi-skills-driven-development-pipeline-stage
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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

1. **Create the transform file**

   For dbt:
   ```
   models/{layer}/{stage_name}.sql
   ```
   ```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
   ```
   ```python
   def run(df: pd.DataFrame) -> pd.DataFrame:
       """Transform {input_table} → {stage_name}."""
       # TODO: add business logic
       return df
   ```

2. **Define the schema contract**
   Create `models/{layer}/schema/{stage_name}.yaml` (dbt) or `pipelines/transforms/{stage_name}/schema.py`:
   ```yaml
   - 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.

3. **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}`.

4. **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

5. **Register in the pipeline DAG**
   Add the stage after its upstream dependencies:
   ```python
   # dags/pipeline.py
   {stage_name}_task = DbtRunOperator(
       task_id="{stage_name}",
       models="{stage_name}",
   )
   {upstream_task} >> {stage_name}_task
   ```

6. **Run locally**
   ```bash
   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.

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [zakelfassi](https://github.com/zakelfassi)
- **Source:** [zakelfassi/skills-driven-development](https://github.com/zakelfassi/skills-driven-development)
- **License:** MIT
- **Homepage:** https://zakelfassi.github.io/skills-driven-development/

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-zakelfassi-skills-driven-development-pipeline-stage
- Seller: https://agentstack.voostack.com/s/zakelfassi
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.