# Bauplan Safe Ingestion

> Ingest data from S3 into Bauplan safely using branch isolation and quality checks before publishing. Use when loading new data from S3, importing parquet/csv/jsonl files, or when the user needs to safely load data with validation before merging to main.

- **Type:** Skill
- **Install:** `agentstack add skill-bauplanlabs-bauplan-skills-bauplan-safe-ingestion`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [BauplanLabs](https://agentstack.voostack.com/s/bauplanlabs)
- **Installs:** 0
- **Category:** [Cloud & Infrastructure](https://agentstack.voostack.com/c/cloud-infrastructure)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [BauplanLabs](https://github.com/BauplanLabs)
- **Source:** https://github.com/BauplanLabs/bauplan-skills/tree/main/plugins/bauplan/skills/bauplan-safe-ingestion

## Install

```sh
agentstack add skill-bauplanlabs-bauplan-skills-bauplan-safe-ingestion
```

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

## About

# Safe Ingestion

Safely ingest data from S3 into the Bauplan lakehouse by isolating changes on a temporary branch, running quality checks, and only merging to `main` after validation succeeds.
This pattern is formally known as Write-Audit-Publish (WAP) in the Iceberg ecosystem.

Implement this as a Python script using the `bauplan` SDK. Do NOT use CLI commands for the ingestion itself.

## Environment Setup

Before writing the script, check whether the project uses `uv` (look for `pyproject.toml` or `uv.lock`). If so, use `uv run python` to execute scripts and `uv add` to install packages. Otherwise, use the system `python` and `pip install`.

Ensure the required packages are installed:
- `bauplan` (the Bauplan Python SDK — required)
- `polars` (if validation logic needs DataFrame operations — zero-copy Arrow interop)

**Do not use pandas.** Bauplan's `client.query()` returns a PyArrow table directly — you can access columns with `result.column("name")[0].as_py()` or convert to Polars with `pl.from_arrow(result)`. No `.to_arrow()` call is needed. Pandas requires a full data copy and is slower.

**The three phases:**
1. **Import** — load data onto a temporary branch (never `main`)
2. **Validate** — run quality checks before publishing
3. **Merge** — promote to `main` only after validation passes

**Branch safety**: All operations happen on a temporary branch, NEVER on `main`. By default, branches are kept open for inspection after success or failure.

**Atomic multi-table operations**: `merge_branch` is atomic. You can create or modify multiple tables on a branch, and when you merge, either all changes apply to main or none do. This enables safe multi-table ingestion workflows.

## Required User Input

Before writing the script, you MUST gather:

1. **S3 path** (required): The S3 URI pattern for the source data (e.g., `s3://bucket/path/*.parquet`)
2. **Table name** (required): The name for the target table
3. **Validation path** (required): See "Choosing a Validation Path" below
4. **On success behavior** (optional):
   - `inspect` (default): Keep the branch open for user inspection before merging
   - `merge`: Automatically merge to main and delete the branch
5. **On failure behavior** (optional):
   - `inspect` (default): Leave the branch open for inspection/debugging
   - `delete`: Delete the failed branch

## Choosing a Validation Path

Ask the user which situation they're in. This determines what validation code goes into the script.

### Path A — "I don't know this data yet"

The user is importing data they haven't explored. They can't state expectations because they don't know the shape yet.

**Action:** Generate the script with a minimal check — table is non-empty (row count > 0). No further questions needed.

**Be honest about what this means:**
- No quality gate protects the merge. If `on_success="merge"`, bad data (wrong types, nulls in key columns, duplicates, stale records) will land on `main` with no way to catch it before downstream consumers read it.
- Quality checks can be added to the script later, but they will not retroactively gate this import or any import run before the checks are added.

**Minimal validation code:**

```python
def validate_import(client, table_name, branch, namespace="bauplan"):
    fq_table = f"{namespace}.{table_name}"
    result = client.query(f"SELECT COUNT(*) as n FROM {fq_table}", ref=branch)
    row_count = result.column("n")[0].as_py()
    assert row_count > 0, f"{fq_table} has 0 rows after import"
    print(f"  Row count: {row_count}")
```

### Path B — "I know what I want to check"

The user can state expectations directly — specific columns, properties, and severities. Examples:

- "user_id must be unique and have no nulls"
- "age must be between 0 and 120"
- "gender must be one of M, F, Non-binary"
- "engagement_score should have no nulls, warn if mean drops below 10"

**Action:** Gather the user's check specifications, then invoke the `bauplan-data-quality-checks` skill to generate a `validate_import()` function. Pass it:

- Table name and branch
- Context: ingestion
- The user's specifications as-is

The `bauplan-data-quality-checks` skill will translate the specifications into validation code. Embed the resulting `validate_import()` function in the script.

**Do not ask about downstream pipelines or consumers.** The user has already decided what to check. If a specification is incomplete (e.g., "check user_id" without saying what property), ask about the specific check, not the pipeline's purpose.

### Path C — "I'll add checks after I build the pipeline"

The user wants to import the data now and build the pipeline first. Once the pipeline exists, the pipeline code will tell the agent exactly what to check.

**Action:** Generate the script with minimal validation (same as Path A). After a successful import, tell the user:

> "The data is imported on branch ``. When your pipeline is ready, you can come back and add quality checks — the `bauplan-data-quality-checks` skill can read your `models.py` and derive checks from how the pipeline actually uses this table."

This is not a failure or a shortcut. It's the right order when the user doesn't yet know what the data's consumers need.

---

## Script Template

```python
"""
Safe ingestion script for .
Write-Audit-Publish (WAP) pattern: import on an isolated branch, validate, then merge or inspect.
"""
import sys
import time
import bauplan

TABLE_NAME = ""
S3_PATH = ""
NAMESPACE = "bauplan"

def validate_import(client, table_name, branch, namespace="bauplan"):
    """Run quality checks on the imported data. Raises on FAIL, prints on WARN."""
    fq_table = f"{namespace}.{table_name}"

    # --- Minimal check: table must be non-empty ---
    result = client.query(f"SELECT COUNT(*) as n FROM {fq_table}", ref=branch)
    row_count = result.column("n")[0].as_py()
    assert row_count > 0, f"{fq_table} has 0 rows after import"
    print(f"  Row count: {row_count}")

    # For Path B: replace the above with checks from the bauplan-data-quality-checks skill.
    # For Path A/C: the above is sufficient.

def main():
    client = bauplan.Client()
    info = client.info()
    username = info.user.username
    timestamp = int(time.time())
    branch_name = f"{username}.import_{TABLE_NAME}_{timestamp}"

    print(f"Creating branch: {branch_name}")
    client.create_branch(branch=branch_name, from_ref="main")

    try:
        # === IMPORT PHASE ===
        print(f"\nPhase 1: Creating table '{TABLE_NAME}' from S3...")
        client.create_table(
            table=TABLE_NAME,
            search_uri=S3_PATH,
            branch=branch_name,
            namespace=NAMESPACE,
            replace=True,
        )
        print(f"  Table schema created.")

        print(f"  Importing data...")
        import_state = client.import_data(
            table=TABLE_NAME,
            search_uri=S3_PATH,
            branch=branch_name,
            namespace=NAMESPACE,
        )
        if import_state.error:
            raise RuntimeError(f"import_data failed: {import_state.error}")
        print(f"  Data imported.")

        # === VALIDATION PHASE ===
        print(f"\nPhase 2: Running quality checks...")
        validate_import(client, TABLE_NAME, branch_name, NAMESPACE)

        # === MERGE PHASE ===
        # on_success="inspect" (default): keep branch open
        print(f"\nImport complete. Branch ready for inspection: '{branch_name}'")
        print(f"To query:  bauplan query \"SELECT * FROM {NAMESPACE}.{TABLE_NAME} LIMIT 10\" --ref {branch_name}")
        print(f"To merge:  bauplan branch merge {branch_name} --into main")
        print(f"To delete: bauplan branch delete {branch_name}")

        # on_success="merge": uncomment below, remove above
        # client.merge_branch(source_ref=branch_name, into_branch="main")
        # print(f"Successfully published {TABLE_NAME} to main")
        # client.delete_branch(branch_name)

    except Exception as exc:
        print(f"\nImport FAILED: {exc}")
        print(f"Branch preserved for debugging: '{branch_name}'")
        sys.exit(1)

if __name__ == "__main__":
    main()
```

## Key SDK Methods

| Method                                         | Description                                           |
|------------------------------------------------|-------------------------------------------------------|
| `bauplan.Client()`                             | Initialize the bauplan client                         |
| `client.info()`                                | Get client info; access username via `.user.username` |
| `client.create_branch(name, from_ref="main")`  | Create a new branch from specified ref                |
| `client.has_branch(name)`                      | Check if branch exists                                |
| `client.delete_branch(name)`                   | Delete a branch                                       |
| `client.create_table(table, search_uri, ...)`  | Create table with schema inferred from S3             |
| `client.import_data(table, search_uri, ...)`   | Import data from S3 into table                        |
| `client.query(query, ref)`                     | Run SQL query, returns a PyArrow Table directly       |
| `client.merge_branch(source_ref, into_branch)` | Merge branch into target                              |
| `client.has_table(table, ref, namespace)`      | Check if table exists on branch                       |

## Workflow Checklist

- [ ] Ask user for: S3 path, table name, on_success, on_failure
- [ ] Ask which validation path: A (don't know data), B (know what to check), or C (will add checks later)
- [ ] Path A or C: write script with minimal validation
- [ ] Path B: gather check specifications, invoke `bauplan-data-quality-checks` skill, embed result in script
- [ ] Run script: `python .py`
- [ ] Verify output shows row count > 0
- [ ] If on_success="inspect": confirm branch ready for review
- [ ] If on_success="merge": confirm merge to main succeeded

## Example Output

**Successful run (on_success="inspect")**:
```
Imported 15234 rows
Import complete. Branch ready for inspection: 'alice.import_orders_1704067200'.
To merge manually: bauplan checkout main && bauplan branch merge alice.import_orders_1704067200
```

**Successful run (on_success="merge")**:
```
Imported 15234 rows
Successfully published orders to main
Cleaned up branch: alice.import_orders_1704067200
```

**Failed run (on_failure="inspect")**:
```
Import failed: No data was imported
Branch preserved for inspection/debugging: 'alice.import_orders_1704067200'
```

## Strengthening Validation Later

If the user chose Path A or C and wants to add checks to an existing script:

1. Invoke the `bauplan-data-quality-checks` skill with context: ingestion.
2. The user provides either:
   - Their own check specifications (they now know the data)
   - A path to `models.py` (they built the pipeline and want checks derived from it)
3. The `bauplan-data-quality-checks` skill reads the existing script, finds the validate phase, and replaces the minimal check with proper validation logic.

Checks added after an import do not gate any previous run. If the data is already on `main`, it's there without quality validation.

## Appending to Existing Tables

To append data to a table that already exists on main, skip `create_table` and only call `import_data`:

```python
# Table already exists on main — just import new data
client.import_data(
    table=table_name,
    search_uri=s3_path,
    namespace=namespace,
    branch=branch_name,
)
```

The validate and merge phases remain the same. New rows are sandboxed on the branch until merged.

## CLI Merge After Inspection

When `on_success="inspect"` (default), the branch is left open for review. To merge after inspecting:

```bash
bauplan checkout main
bauplan branch merge 
bauplan branch rm   # optional cleanup
```

The branch name is printed by the script upon completion.

---

## Reference

When unsure about a method signature, CLI flag, or concept, fetch the relevant doc page via `WebFetch` rather than guessing. Pages are markdown and LLM-friendly.

**Python SDK:** `https://docs.bauplanlabs.com/reference/bauplan.md`

**Relevant guides and concept pages:**
- Import data: `https://docs.bauplanlabs.com/tutorial/import.md`
- Schema conflicts: `https://docs.bauplanlabs.com/concepts/schema-conflicts.md`
- Handle casting programmatically: `https://docs.bauplanlabs.com/concepts/schema-conflicts.md`
- Data branches: `https://docs.bauplanlabs.com/concepts/git-for-data/data-branches.md`
- Tables: `https://docs.bauplanlabs.com/concepts/tables.md`
- Namespaces: `https://docs.bauplanlabs.com/concepts/namespaces.md`

**Full doc index:** `https://docs.bauplanlabs.com/llms.txt`

**CLI:** The `bauplan` CLI is self-documenting:
- `bauplan --help` — lists all available commands
- `bauplan  --help` — shows arguments and options for a specific command (e.g., `bauplan branch --help`, `bauplan import-data --help`)

**Validating generated Python:** After writing or updating the ingestion script, run `ruff check` and `ruff format` to catch syntax errors and style issues, and `ty` to catch type errors — these verify the code compiles and the SDK calls are well-formed without executing it. Only run these if they are installed (check with `which ruff` / `which ty`).

## Source & license

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

- **Author:** [BauplanLabs](https://github.com/BauplanLabs)
- **Source:** [BauplanLabs/bauplan-skills](https://github.com/BauplanLabs/bauplan-skills)
- **License:** MIT

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-bauplanlabs-bauplan-skills-bauplan-safe-ingestion
- Seller: https://agentstack.voostack.com/s/bauplanlabs
- 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%.