# Factory Data Pipeline Engineer

> Use when designing or implementing data ingestion, CSV imports, time-series storage, Python services that sit alongside Next.js, simulation pipelines, or external-API integration with submit/poll/fetch shapes. Carries the factory's data-pipeline conventions — TS scripts with Papa Parse, JSONB envelopes for time-series, the three-entry-point Python pattern (CLI / Cloud Run API / Pub/Sub), YAML config for service-level data, converter/service split, Cloud Run + ephemeral Neon deployment.

- Skill: `nonlinear-xyz/factory-data-pipeline-engineer` (Agent Skill)
- Install (CLI): `npx skillmds@latest add nonlinear-xyz/factory-data-pipeline-engineer`
- Raw SKILL.md: https://api.skillmd.com/api/skills/nonlinear-xyz/factory-data-pipeline-engineer/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Web & Frontend
- Author: nonlinear-xyz (https://skillmd.com/u/nonlinear-xyz)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/nonlinear-xyz/factory-data-pipeline-engineer

---


Apply the **data-pipeline-engineer** specialist workflow. Design data flow that fits the factory's pipeline conventions, not bespoke ETL plumbing. Load `factory-data-pipelines` and `factory-stack` through the host's skill capability when needed.

## How to think (in order)

1. **What's the data shape?** Pick one:
   - **One-shot or scheduled CSV** → TS script in `scripts/data_processing/`
   - **Event stream / time-series** → JSONB envelope on a structured parent table
   - **External API ingestion** (slow operation) → submit/poll/fetch async pattern
   - **Reference data (slowly changing)** → YAML config (Python side)
   - **Compute job** (sim, optimization, ML) → Python service with three entry points

   If it doesn't match one, that's the finding — name it.

2. **TS or Python?** Default: Next.js side (TS). Move to Python when:
   - Numeric / scientific libraries are non-trivial (geopandas, shapely, numpy/scipy)
   - Existing Python expertise / models
   - Compute runtime > Vercel function timeout (~10s on hobby, 60s on pro)

3. **Storage shape?** Drizzle table with structured columns for **what drives queries** + JSONB for **what doesn't**. Rule: if you need to filter or sort by it at app speed, it earns a column.

4. **Deployment shape?**
   - **TS script** → run locally or in GitHub Action; commits the data to DB
   - **Cloud Run API** → HTTP endpoint, FastAPI, API key dependency
   - **Cloud Run Pub/Sub handler** → async job processor
   - **Long-running compute** → Cloud Run with extended timeout, or Cloud Run Jobs

5. **Migrations?** Run in CI, not at runtime. Drizzle's generate + push, or dbmate for raw SQL projects.

6. **Idempotency?** Default to upsert-on-conflict for CSV imports. Wrap in a transaction. Don't assume "imported once."

7. **Converter vs service split?** Pure transforms in `*-converter.ts` (client-safe). I/O in `*-service.ts` (server-only). Don't blur.

## Reference: canonical TS import script

```ts
// scripts/data_processing/import-foo.ts
import { readFileSync } from 'fs';
import Papa from 'papaparse';
import { db } from '@/db';
import { foo } from '@/db/schema';

const csvText = readFileSync(process.argv[2], 'utf8');
const { data, errors } = Papa.parse<FooRow>(csvText, {
  header: true,
  skipEmptyLines: true,
  dynamicTyping: true,
});

if (errors.length) {
  console.error('Parse errors:', errors);
  process.exit(1);
}

await db.transaction(async (tx) => {
  for (const row of data) {
    await tx.insert(foo).values({
      externalId: row.external_id,
      name: row.name,
      // ... map every column
    }).onConflictDoUpdate({
      target: foo.externalId,
      set: { name: row.name, updatedAt: new Date() },
    });
  }
});

console.log(`Imported ${data.length} rows`);
```

## Reference: canonical Python service layout

```
models/<service>/
├── Dockerfile             # Pub/Sub variant
├── Dockerfile.api         # API variant
├── pyproject.toml
├── main.py                # CLI entry
├── main_api.py            # FastAPI entry — Cloud Run HTTP
├── main_pubsub.py         # Pub/Sub handler entry
├── config/
│   ├── routes.yaml
│   └── vehicles.yaml
├── src/
│   ├── api/
│   │   └── deps.py        # API key, request context
│   ├── models/
│   │   ├── request.py     # Pydantic
│   │   └── message.py     # Pub/Sub message shape
│   ├── services/
│   │   ├── data_service.py    # YAML loader
│   │   ├── mapbox_service.py
│   │   └── gcs_service.py
│   └── simulation/
│       └── simulation_runner.py   # core work; mode-aware
└── tests/
    └── conftest.py
```

## Output format

```
## Restated request
<one sentence>

## Pipeline shape
- Data shape: <CSV / event stream / external API / reference data / compute>
- Runtime: <TS / Python / mixed — why>
- Storage: <columns / JSONB envelope / both>
- Deploy: <local script / Vercel scripts / Cloud Run API / Cloud Run Pub/Sub>

## Files to create or modify
<bulleted list with paths>

## Code
<organized by file>

## Operational details
- Idempotency: <how>
- Migrations: <CI step, not runtime>
- Trace ID: <yes — middleware in place>
- Logging: <PostHog / Sentry / structured logs>

## Open questions
<things the user should confirm>
```

## What you do NOT do

- **Don't pre-build `libs/py-libs/` shared utilities.** Wait for the second consumer.
- **Don't query inside JSONB at app speed.** Promote the field to a column.
- **Don't write a Cloud Run job before the second use case.** Standalone TS script is enough until proven.
- **Don't share Pydantic models across the three Python entry points by copy-paste.** Define once in `models/`, import everywhere.
- **Don't run migrations at runtime.** CI's job.
- **Don't blur converter / service boundaries.** Pure-transform stays in `*-converter.ts`; I/O in `*-service.ts`.
- **Don't write Drizzle in raw `pg` style.** If you're considering raw SQL with row-mappers, you're doing it wrong.
- **Don't put YAML paths in hardcoded strings.** Env-driven base paths.

## When the request is too small for this framework

If the user asks for a one-off query, a single Drizzle insert, or "just read this CSV once," do it directly without the full pipeline framework. The framework is for recurring or productionized data flow.