Land

land — Generate and Run a dlt Raw-Landing Pipeline

Overview

This skill generates a dlt pipeline script from the endpoints.json produced by assess, then runs it to land raw data in the chosen destination. No transformation logic is added here. Transformation (dbt, modeling) is out of scope for this plugin and belongs in a separate downstream pipeline.

Pre-requisite: endpoints.json must exist in the project root. Run assess first if it is missing.

---

Step 1 — Choose a Destination

Ask:

Which destination should we land the data into? Common choices: DuckDB (local file, good for dev), BigQuery, Snowflake, Redshift, Postgres. A files/blob destination (Parquet or CSV via dlt's filesystem destination) is also supported — see ../../references/destinations.md for the dlt.destinations.filesystem call and the environment variables it requires.

See references/destinations.md for the exact dlt.destinations.* call and the environment variables each destination requires.

Wait for the user's answer before continuing.

---

Step 2 — Confirm Secrets

Check references/security.md for the security rules. Then ask:

Please confirm that all required secrets (API tokens, warehouse credentials) are set as environment variables in your shell before we generate the script — do not paste them here.

For example:

export MY_API_TOKEN="sk-..."

Confirm the variables are set, then we'll proceed.

Do not proceed until the user confirms.

---

Step 3 — Generate the Pipeline Script

Read endpoints.json to inspect auth.type. Generate the appropriate variant below and save it as pipeline_land.py in the project root.

Variant A — No Auth

"""Raw-landing pipeline — generated by api-warehouse land skill."""
import json
import dlt
from dlt.sources.rest_api import rest_api_source
from api_warehouse.pipeline import build_rest_api_config

# Load the spec produced by `assess`
with open("endpoints.json") as f:
    spec = json.load(f)

# Build the dlt rest_api config (no auth)
config = build_rest_api_config(spec)
source = rest_api_source(config)

# Run into destination — RAW LANDING ONLY, no transformation
pipeline = dlt.pipeline(
    pipeline_name="api_warehouse_land",
    destination=dlt.destinations.duckdb("warehouse.duckdb"),  # change as needed
    dataset_name="raw",
)
load_info = pipeline.run(source)
print(load_info)
# Rows loaded per resource (verifiable by eval loop):
try:
    for table, count in pipeline.last_trace.last_normalize_info.row_counts.items():
        if not table.startswith("_dlt"):
            print(f"  {table}: {count} rows")
except Exception:
    pass

Variant B — Bearer Token Auth

"""Raw-landing pipeline — generated by api-warehouse land skill."""
import json
import os
import dlt
from dlt.sources.rest_api import rest_api_source
from api_warehouse.pipeline import build_rest_api_config

# Load the spec produced by `assess`
with open("endpoints.json") as f:
    spec = json.load(f)

# Read the secret from the environment — never hard-code tokens
token_env = spec["auth"]["token_env"]          # e.g. "MY_API_TOKEN"
secrets = {token_env: os.environ[token_env]}

# Build the dlt rest_api config with bearer auth
config = build_rest_api_config(spec, secrets)
source = rest_api_source(config)

# Run into destination — RAW LANDING ONLY, no transformation
pipeline = dlt.pipeline(
    pipeline_name="api_warehouse_land",
    destination=dlt.destinations.duckdb("warehouse.duckdb"),  # change as needed
    dataset_name="raw",
)
load_info = pipeline.run(source)
print(load_info)
# Rows loaded per resource (verifiable by eval loop):
try:
    for table, count in pipeline.last_trace.last_normalize_info.row_counts.items():
        if not table.startswith("_dlt"):
            print(f"  {table}: {count} rows")
except Exception:
    pass

Adapt the dlt.destinations.* call to match the user's chosen destination (see references/destinations.md for BigQuery, Snowflake, Redshift, Postgres variants).

Check MEMORY.md for any destination-specific or API-specific quirks before writing the final script.

Show the full generated script to the user and ask them to confirm before running.

---

Step 4 — Self-check (Evals)

Before running the script, spin up a grader agent with a clean context. Give it EVALS.md and the generated pipeline_land.py. Follow the loop defined in ../../references/running-evals.md. Fix any fail verdicts before proceeding. On platforms without subagents (e.g. Codex), run the same checklist inline in a fresh reasoning pass instead — see ../../references/running-evals.md.

---

Step 5 — Run the Script

Once evals pass and the user confirms, run:

python pipeline_land.py

Parse load_info to report rows loaded per resource:

Resource: posts       rows loaded: 100
Resource: comments    rows loaded: 500

If the run fails:

• Auth errors: re-check that env vars are set and match endpoints.json token_env.
• Destination errors: re-check credentials per references/destinations.md.
• Schema errors: share the traceback with the user and suggest re-running assess.

For any user-facing text in the report, apply the checks in ../../references/anti-slop.md — cut filler, keep field names and row counts.

---

Step 6 — Checkpoint

Tell the user:

Raw landing complete. Data is in the raw dataset in your destination. Resource row counts above. Next step: run validate to reconcile counts against the source API and confirm completeness.

Do not run validate automatically — wait for the user to proceed.