# Land

> Use whenever the user wants to load/extract/ingest API data into a warehouse or blob, build a dlt pipeline, or land raw data into BigQuery, Snowflake, Postgres, Azure, or local files — even if they don't say "land". Trigger phrases: "load raw data", "build a dlt pipeline", "ingest into DuckDB", "land the data", "run the pipeline", "extract API data", or any request to move API data into a destination table.

- Skill: `sdhilip200/land` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds@latest add sdhilip200/land`
- Raw SKILL.md: https://api.skillmd.com/api/skills/sdhilip200/land/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: sdhilip200 (https://skillmd.com/u/sdhilip200)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/sdhilip200/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:
> ```bash
> 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

```python
"""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

```python
"""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:

```bash
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.