# Build a data pipeline with Cursor URL: https://developertoolkit.ai/en/cursor-ide/build/data-pipeline/ Description: Build an ETL pipeline with Cursor Agent: Plan Mode contract, fixture tests guarded against Agent edits, idempotent loads, and reconciliation proving the data. Cursor builds a trustworthy data pipeline when Plan Mode first writes a contract (per-stage schemas, natural keys, watermarks and invariants) and Agent then implements extract, transform and load against fixture tests it may not edit. Re-run idempotency tests, row and sum reconciliation against the source, and post-load quality checks prove the data correct without anyone reading every generated line. This guide is for developers and data engineers who build pipelines in Cursor, and for tech leads who have to sign off on the numbers. The situation: analytics needs a daily report built from a Postgres replica, the Stripe API and CSV files a warehouse system drops into S3. Today a 400-line script does the job. It fails silently when a CSV column is renamed, duplicates rows when someone re-runs it, and at 3 AM nobody knows which stage stopped. Agent can draft the replacement quickly. What it cannot do on its own is know that an inner join which silently drops 2% of orders is a bug, so the work of this page is making that knowable. ## What you'll walk away with from building a pipeline in Cursor - A Plan Mode prompt that turns requirements into a **pipeline contract** file Agent implements against. - A project rule that makes idempotency, quarantine files and structured errors the default in every stage. - Read-only database access for Agent through a Postgres MCP server, so schemas come from the source rather than from guesses. - Six copy-paste prompts for extractors, test-first transforms, an idempotent loader, orchestration and quality checks. - A verification ladder that proves a run correct from counts and sums, plus a failure-mode table with recovery steps. ## Why does a pipeline need a contract before code? Agent generates plausible transformation code quickly, and plausible is the problem: a pipeline that is wrong still produces tables, and dashboards read those tables without complaint. The contract is the oracle. It names, for each stage, the input and output schema, the natural key, the incremental watermark, and the invariants that must hold, such as "every extracted order is either loaded or quarantined". Once those invariants exist as a file, tests and checks can enforce them, and review becomes a question of whether the contract is right instead of whether 2,000 lines of TypeScript look right. This guide applies the principle of [verifying evidence instead of reading diffs](/en/ladder/evidence-not-diffs/). ## Step 1: Write the pipeline contract in Plan Mode Plan Mode creates an implementation plan before Agent writes any code (see [Agent modes in depth](/en/cursor-ide/build/agent-modes-deep-dive/)). Use it to produce the contract, then review and commit that file before the first line of pipeline code exists. **Copy-paste prompt: pipeline contract in Plan Mode** ```text Plan a daily batch pipeline and write the result to docs/pipeline-contract.md. Do not write any pipeline code yet. Sources: - Postgres read replica: orders, order_items, customers, products (use updated_at for incremental reads) - Stripe API: payment intents, refunds, disputes for the run date - CSV inventory snapshots in s3://acme-warehouse-exports/inventory/YYYY-MM-DD/ Destination: Snowflake, schema ANALYTICS, star schema (fact_orders, dim_customer, dim_product, fact_inventory). Requirements: - Runs daily at 02:00 UTC for the previous UTC day; supports backfill of a date range. - Idempotent: re-running any date produces identical tables, never duplicates. - Partial failure: if Stripe is down, load the other sources and mark the run "partial". - Every input row ends up loaded or quarantined with a reason. Nothing disappears. For each stage (extract per source, transform, load, quality checks) write: 1. Input and output schema with types and nullability. 2. Natural key and deduplication rule. 3. Watermark or partition used for incremental and re-run behaviour. 4. Invariants a test can check (row conservation, sum conservation of amounts, key uniqueness). 5. Failure behaviour: retry, skip-and-mark-partial, or abort. Before writing the file, ask me about anything you would otherwise have to assume: volumes, late-arriving data, currency handling, and timezone of the source timestamps. ``` Read the contract, not the code that will follow. The questions worth a human's time are all in it: is `order_id` really unique across re-extracts, are Stripe amounts in the smallest currency unit, and does "previous day" mean UTC or the business timezone. A wrong answer here becomes a wrong dashboard later, and it is cheaper to fix in a 150-line Markdown file. ## Step 2: Add a project rule for pipeline code A [project rule](/en/cursor-ide/quick-start/project-rules/) carries the conventions into every Agent session, so the fifth extractor follows the same pattern as the first. Scope it to the pipeline directories and keep it short; the file format and scoping options are on the project rules page. ```text Pipeline code in src/extractors, src/transformers, src/loaders, src/quality: - The contract in docs/pipeline-contract.md is the source of truth. If code and contract disagree, stop and ask; do not change the contract to fit the code. - Every stage is idempotent for a given run_date. Loads use MERGE on the natural key, never INSERT. - Validate every external payload with a Zod schema at the boundary. Unknown fields are logged and dropped, not fatal. Missing required fields send the row to the quarantine file with a reason. - Every stage returns { rowsIn, rowsOut, rowsQuarantined } and asserts rowsIn = rowsOut + rowsQuarantined. - Every error log includes run_id, stage, source, run_date and the offending key. - Timestamps are handled in UTC; convert only in the presentation layer. - Never edit files under tests/fixtures or tests/expected. If a fixture test fails, fix the code or report why the expected output is wrong. ``` The last line matters more than it looks. An agent under pressure to make a test pass will sometimes "fix" the expected output instead of the code. [Protecting the oracle](/en/workflows/test/protect-the-oracle/) covers the full set of defences; for a pipeline, the minimum is this rule plus a CI check that fails when a fixture changes in the same pull request as the code it tests. This GitHub Actions workflow is that check. It needs the full history (`fetch-depth: 0`) to diff against the base branch, and read-only repository access: ```yaml title=".github/workflows/fixture-guard.yml" name: fixture-guard on: pull_request permissions: contents: read jobs: guard: runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 with: persist-credentials: false fetch-depth: 0 - name: Fail when fixtures and pipeline code change together env: BASE_REF: ${{ github.base_ref }} run: | changed=$(git diff --name-only "origin/$BASE_REF...HEAD") if echo "$changed" | grep -q '^tests/\(fixtures\|expected\)/' && echo "$changed" | grep -q '^src/'; then echo "::error::Fixtures and code changed in one pull request. Split them." exit 1 fi ``` A fixture that really is wrong then goes in its own pull request, where the data owner reviews it without code changes to hide behind. If you prefer an approval gate to a hard failure, add `tests/fixtures/ @data-owners` to `CODEOWNERS` and turn on required code-owner review for the branch. Either way, the check alone is not tamper-proof: on `pull_request`, GitHub runs this workflow file from the pull request itself, so a pull request can edit `fixture-guard.yml` and still go green. Add `/.github/ @platform-owners` to `CODEOWNERS` with required code-owner review, and make `fixture-guard` a required status check, so a change to the guard cannot merge without its owner. ## Step 3: Give Agent read-only access to the source schema Agent writes better extractors when it can read the real column types, nullability and indexes instead of inferring them from old code. Postgres MCP Pro (PyPI `postgres-mcp`, 0.3.0, last released 2025-05-16, checked 2026-09-26) does this, and `--access-mode=restricted` limits it to read-only transactions with execution-time limits. It has had no release in over a year, so pin the version and prefer DBHub or MCP Toolbox from [Database MCP servers](/en/ecosystem/mcp/database-mcp/) if you need an actively maintained server. It is a Python package, so it runs through `uvx` (which requires [uv](https://docs.astral.sh/uv/)), not `npx`. Put the entry in the global `~/.cursor/mcp.json`, not the project's `.cursor/mcp.json`, so the connection string stays out of the repository: ```json title="~/.cursor/mcp.json" { "mcpServers": { "postgres": { "command": "uvx", "args": ["postgres-mcp==0.3.0", "--access-mode=restricted"], "env": { "DATABASE_URI": "postgresql://agent_ro@replica.internal:5432/app" } } } } ``` Keep the password for the `agent_ro` role in `~/.pgpass` rather than in the JSON file, and grant that role `SELECT` only. Point it at a replica or a staging copy, never at the primary: rows the agent reads reach the model, and a free-text column can carry instructions. [Database MCP servers](/en/ecosystem/mcp/database-mcp/) compares the alternatives (DBHub, MCP Toolbox, Supabase, Neon), and [MCP setup in Cursor](/en/cursor-ide/quick-start/mcp-setup/) covers enabling and debugging a server. ## Step 4: Build the extractors with quarantine, not silent drops Each extractor produces the same shape: JSONL files in a staging directory plus a quarantine file and a counts record. Build one extractor per session so each diff stays reviewable by its tests. **Copy-paste prompt: incremental Postgres and Stripe extractors** ```text Implement the Postgres and Stripe extractors described in docs/pipeline-contract.md. Postgres (src/extractors/postgres.ts): - Use the postgres MCP server to read the actual column types of orders, order_items, customers and products before writing any types. Generate TypeScript row types from them. - Use pg with a Pool. Read rows where updated_at is in [watermark, run_date + 1 day). - Store the watermark per table in the pipeline_state table only after the stage succeeds. Stripe (src/extractors/stripe.ts): - Use the stripe SDK's auto-pagination for payment intents, refunds and disputes created in the run date. - Retry 429 and 5xx with exponential backoff and jitter, max 5 attempts. If Stripe stays unavailable, return status "skipped" with the error instead of throwing. Both extractors: - Validate every row with a Zod schema. Invalid rows go to staging/quarantine/.jsonl with the validation error; valid rows go to staging//.jsonl. - Return { rowsIn, rowsOut, rowsQuarantined } and assert rowsIn = rowsOut + rowsQuarantined. Write Vitest tests with recorded fixtures (no network): a page boundary in Stripe pagination, a 429 followed by success, a row with a null required field, and a re-run with an unchanged watermark that must extract zero new rows. Run the tests and show me the output. ``` CSV sources deserve their own prompt because they change without notice. Ask Agent to map header names through a config file (`config/csv-columns.json`) so a renamed column is a one-line config change, to treat a missing daily file as "no data" rather than an error when the contract says weekends are skipped, and to parse with `csv-parse` so quoted fields containing newlines survive. The quarantine rule from Step 2 applies unchanged. ## Step 5: Write the transformation tests before the transformations Transformations are where silent data loss happens: a join on a nullable key, a deduplication that keeps the older row, a cents-to-dollars conversion applied twice. Make Agent write the failing tests first, from the contract, and only then the code. **Copy-paste prompt: test-first order transformations** ```text Read docs/pipeline-contract.md, section "transform". Do not write src/transformers/orders.ts yet. First, write tests in tests/transformers/orders.test.ts with small hand-written fixtures in tests/fixtures/orders/ covering: 1. An order with no matching customer (customer name "Unknown", email null, row kept). 2. An order with no Stripe payment (payment_status "pending", row kept). 3. The same order_id in today's and yesterday's extract (keep the row with the later updated_at). 4. Amounts in cents converted to dollars exactly once, using integer arithmetic until the final value. 5. An order whose updated_at is 23:59:59.999 UTC lands in the correct run_date. Then add property-based tests with fast-check over generated inputs: - Row conservation: output rows + quarantined rows = distinct input order_ids. - Sum conservation: the sum of amount_cents in equals the sum out plus the sum quarantined. - Order independence: shuffling the input rows does not change the output. Run the tests, confirm they fail for the right reason, then implement the transformer until they pass. Do not modify anything under tests/fixtures after this point. ``` The property tests carry most of the weight, because they check an invariant over thousands of generated inputs instead of the five cases you thought of. [Property-based testing with agents](/en/workflows/test/property-based-testing/) goes deeper on choosing properties; [test data management](/en/workflows/test/test-data-management/) covers building realistic fixtures without copying production rows. ## Step 6: Make the loader idempotent and prove it by running it twice **Copy-paste prompt: idempotent Snowflake loader** ```text Implement src/loaders/snowflake.ts per docs/pipeline-contract.md, section "load". - Use snowflake-sdk. Load each JSONL file into a temporary staging table in 10,000-row batches. - Then run one MERGE per table from the staging table into the target, on the natural key, inside a transaction: update when the source updated_at is newer, insert when missing. On failure roll back and log run_id, table and the error. - Write a pipeline_runs row: run_id, run_date, status, and rowsIn/rowsOut/rowsQuarantined per stage. - Support --dry-run: validate and report counts, write nothing. Add an integration test that runs against a disposable schema named PIPELINE_TEST_: load the fixture day, record row counts and SUM(amount_usd) per table, load the same day again, and assert both numbers are unchanged. Drop the schema at the end of the test. Read Snowflake credentials from environment variables only. ``` The load-twice test is the single most valuable test in the pipeline. An `INSERT` where a `MERGE` belongs passes every other check on the first run and doubles revenue on the second. ## Step 7: Orchestrate the run and check the data after loading **Copy-paste prompt: orchestrator and post-load quality checks** ```text Implement src/orchestrator.ts and src/quality/checks.ts per docs/pipeline-contract.md. Orchestrator: - CLI: npx tsx src/orchestrator.ts --date 2026-09-25, or --from and --to for a backfill (one date at a time). - Run the three extractors concurrently, then transform, then load. - A skipped source marks the run "partial" and records which source to retry; any other failure aborts the run before loading and leaves the previous day's tables untouched. - Take a lock row in pipeline_runs for run_date so two runs for the same date cannot overlap. Quality checks (run after load, results written to pipeline_quality): - Reconciliation: orders loaded for run_date equals orders in the Postgres replica for run_date minus quarantined rows; the same for SUM(amount_cents) against Stripe. - Uniqueness of natural keys in every fact table. - Referential integrity: every fact_orders.customer_key exists in dim_customer. - Volume: row count within the range of the previous 28 days' min and max; flag, do not fail. - Freshness: newest updated_at in fact_orders is within 26 hours of the run time. Failures post to the webhook in QUALITY_WEBHOOK_URL with run_id and the failing query. ``` Reconciliation is the check that makes the rest trustworthy. It compares the warehouse against the sources with numbers anyone can recompute, so a reviewer who never opens `orders.ts` can still say whether the day's data is right. For dashboards and alert routing on top of `pipeline_quality`, see [monitoring with Cursor](/en/cursor-ide/operate/monitoring/). ## How do you verify a pipeline without reading every line? Stack the gates so each catches what the previous one cannot, and let the evidence, not the diff, decide the merge. | Gate | What it proves | Runs where | Who looks at the result | | --- | --- | --- | --- | | Type check and lint | Types match the schemas read from the source | CI on every pull request | Nobody, unless red | | Fixture tests (Step 5) | Known edge cases behave as the contract says | CI | Author, on failure | | Property tests (Step 5) | Row and sum conservation over generated inputs | CI | Author, on failure | | Load-twice test (Step 6) | Re-runs do not duplicate or change data | CI against a disposable schema | Author, on failure | | Replay of a real day | Code handles production-shaped data | Staging, before merge | Reviewer reads the reconciliation report | | Post-load checks (Step 7) | Each production run matches its sources | Every scheduled run | On-call, through the webhook | For the replay, run the new code for one recent date into a staging schema and ask Agent to produce a reconciliation report: per table, source count, loaded count, quarantined count, and amount sums, next to the same numbers from the current production pipeline. Attach it to the pull request. That report, the contract diff and green CI are the [evidence bundle](/en/workflows/ship/evidence-bundle/) for the change. The data owner signs off on the contract and the reconciliation numbers; the reviewer confirms fixtures did not change alongside the code. ## Where Cursor-built pipelines break, and how to recover | Symptom | Likely cause | Recovery | | --- | --- | --- | | Loaded counts are a few percent below source, no error | Inner join on a nullable key, or deduplication on the wrong key | Find the lost keys with an anti-join between staging and target. Add a fixture with that key shape, fix the transform, and re-run the affected dates; the MERGE makes the re-run safe. | | Revenue doubles after a manual re-run | Loader uses `INSERT`, or the MERGE key is not the natural key | Delete the run_date partition, fix the loader, and re-run. Add the load-twice test before merging the fix. | | Extractor starts failing on a date with no code change | Source schema drift: a new enum value, a renamed CSV column | Read the quarantine file first. Update the Zod schema or `config/csv-columns.json`, then re-run that date. | | A test went green after Agent's last change, but the data is wrong | Agent edited a fixture or expected output to match the code | Run `git diff --stat -- tests/fixtures tests/expected`, revert the fixture change, and re-prompt with the failing test named. Add the CI check from Step 2. | | Orders from late evening appear on the wrong day | Source timestamps in local time, run_date in UTC | Decide the timezone in the contract, add the 23:59:59.999 fixture, and backfill the affected range. | | A backfill hits Stripe 429s or slows the replica | Backfill runs dates in parallel | Run one date at a time with the orchestrator's `--from`/`--to` loop and keep the backoff settings from Step 4. | | A 3 AM failure says only `Error: undefined` | Errors thrown without context | Ask Agent to audit every `throw` and log against the rule in Step 2, then add a test that forces each stage to fail and asserts the log fields. | Do not let Agent run the pipeline against production Snowflake from a chat session to "check it works". Use a disposable schema and credentials scoped to it; production runs belong to the scheduler, with the post-load checks watching them. ## Where to go next after building your pipeline Working in another tool? The same database work is covered in [Database work in Claude Code](/en/claude-code/build/database-work/) and [Database work with Codex](/en/codex/build/database-work/), and the cross-tool practice is in [Database design and queries](/en/workflows/build/database-development/). - [Database design with Cursor](/en/cursor-ide/plan/database-design/) — Design the warehouse and source schemas the pipeline contract depends on. - [Testing and TDD with Cursor](/en/cursor-ide/test/testing-tdd/) — Run the test-first loop from Step 5 on the rest of your codebase. - [Monitoring with Cursor](/en/cursor-ide/operate/monitoring/) — Turn the post-load checks into dashboards and alerts. - [DevOps with Cursor](/en/cursor-ide/ship/devops/) — Deploy and schedule the pipeline with infrastructure as code. - [Database design and queries (all tools)](/en/workflows/build/database-development/) — The cross-tool workflow for schemas, migrations and queries with any agent. - [Data engineering recipes](/en/cookbook/data-recipes/) — dbt, SQL pipelines, notebooks and data tests with an agent.