API to PostgreSQL Pipeline

Production-minded Python template for ingesting paginated API data into PostgreSQL with incremental loads, idempotent upserts, validation, Docker, tests, and CI.

What This Includes

• HTTP API extraction with retries and pagination
• Pydantic validation before loading
• Raw landing table for auditability
• Curated PostgreSQL model with idempotent ON CONFLICT upserts
• Incremental cursor persisted in Postgres
• Load run tracking and structured JSON logs
• Docker Compose for local Postgres + pipeline execution
• Pytest unit tests
• GitHub Actions CI

Quick Start

cp .env.example .env
docker compose up -d postgres
docker compose run --rm migrate
docker compose run --rm pipeline run

The default example ingests posts from https://jsonplaceholder.typicode.com/posts.

Local Python

python -m venv .venv
. .venv/Scripts/activate
pip install -e ".[dev]"
cp .env.example .env
docker compose up -d postgres
python -m api_to_postgres_pipeline.cli migrate
python -m api_to_postgres_pipeline.cli run

Configuration

Environment variables are defined in .env.example.

Important fields:

• DATABASE_URL: PostgreSQL connection string.
• API_BASE_URL: API host.
• API_ENDPOINT: endpoint path to ingest.
• API_PAGE_PARAM: page query parameter name.
• API_PAGE_SIZE_PARAM: page size query parameter name.
• API_PAGE_SIZE: requested page size.
• API_MAX_PAGES: safety limit for one run.
• SOURCE_NAME: logical source key used for cursor state.

Architecture

Data flow:

API -> validation -> raw.api_records -> public.api_items
                       |
                       v
                 ops.ingestion_state
                 ops.load_runs

The curated model is intentionally simple and can be adapted to a real domain table. The idempotency key is (source_name, source_record_id), and every row stores a deterministic payload hash.

Tests

pytest