Automated API to PostgreSQL — Medallion Data Pipeline

A production-grade data engineering pipeline that ingests product data from a REST API and processes it through a Bronze → Silver → Gold Medallion Architecture using Python and PostgreSQL.

---

Architecture Overview

DummyJSON API
      │
      ▼

   BRONZE      Raw JSON stored as-is — append only
  (Ingestion)  Deduplication via ON CONFLICT

       │
       ▼

   SILVER      Cleaned, typed, normalized into relational tables
 (Transform)   products / tags / images / reviews

       │
       ▼

    GOLD      Aggregated, business-ready views
   (Serve)    category_summary / top_rated / review_summary

---

Project Structure

project/
├── config/
│   └── settings.py               # Environment variable loader
├── migrations/
│   ├── V1__create_schemas.sql    # bronze / silver / gold schemas
│   ├── V2__create_bronze_tables.sql
│   ├── V3__create_silver_tables.sql
│   └── V4__create_gold_tables.sql
├── models/
│   ├── bronze.py                 # Pydantic models
│   ├── silver.py
│   └── gold.py
├── pipelines/
│   ├── ingest/
│   │   └── products_api.py       # Fetch API → Bronze
│   └── transform/
│       ├── bronze_to_silver.py   # Bronze → Silver
│       └── silver_to_gold.py     # Silver → Gold
├── utils/
│   ├── db.py                     # PostgreSQL connection
│   └── logger.py                 # Centralized logging
├── tests/
│   ├── test_ingest.py
│   ├── test_bronze_to_silver.py
│   └── test_silver_to_gold.py
├── .env.example
├── .gitignore
├── requirements.txt
└── README.md

---

Database Schema

Bronze Layer

Column	Type	Description
bronze_id	BIGSERIAL	Auto-increment primary key
ingestion_id	TEXT	UUID per record
source_system	TEXT	Source identifier
source_url	TEXT	API endpoint
ingested_at	TIMESTAMPTZ	Ingestion timestamp
raw_payload	JSONB	Full raw JSON
batch_id	TEXT	UUID per pipeline run

Silver Layer

• silver.products — Cleaned product fields + dimensions
• silver.product_tags — Normalized tag rows
• silver.product_images — Normalized image URLs
• silver.product_reviews — Review records with timestamps

Gold Layer

• gold.category_summary — Aggregated stats per category
• gold.top_rated_products — Highest rated products
• gold.review_summary — Review counts and averages per product

---

Getting Started

Prerequisites

• Python 3.10+
• PostgreSQL 14+
• pip

Installation

git clone https://github.com/your-username/automated-api-to-postgresql.git
cd automated-api-to-postgresql/project
pip install -r requirements.txt

Environment Setup

cp .env.example .env

Edit .env:

PG_HOST=localhost
PG_PORT=5432
PG_DB=your_database
PG_USER=postgres
PG_PASSWORD=your_password

PRODUCTS_API_URL=https://dummyjson.com/products?limit=0
PRODUCTS_API_TIMEOUT=10

Run Migrations

Run SQL files in order in pgAdmin or psql:

V1__create_schemas.sql
V2__create_bronze_tables.sql
V3__create_silver_tables.sql
V4__create_gold_tables.sql

---

Running the Pipeline

Step 1 — Ingest (API → Bronze)

py -m pipelines.ingest.products_api

Step 2 — Transform (Bronze → Silver)

py -m pipelines.transform.bronze_to_silver

Step 3 — Serve (Silver → Gold)

py -m pipelines.transform.silver_to_gold

---

Key Design Decisions

Deduplication

Bronze uses ON CONFLICT ((raw_payload->>'id')) DO NOTHING backed by a unique index, preventing duplicate products across pipeline runs.

Schema Versioning

All DDL lives in numbered migration files — never inside pipeline code. This keeps schema changes versioned, reviewable, and environment-safe.

ELT over ETL

Transformation happens inside PostgreSQL using SQL (INSERT INTO ... SELECT), not in Python. This leverages the database engine for performance and keeps pipeline code simple.

Separation of Concerns

products_api.py      →  reads API,   writes Bronze only
bronze_to_silver.py  →  reads Bronze, writes Silver only
silver_to_gold.py    →  reads Silver, writes Gold only

---

Logging

All pipeline runs log to logs/pipeline.log and stdout:

2026-04-13 14:31:23 - products_api      - INFO - Products fetched successfully. Count: 194
2026-04-13 14:31:24 - products_api      - INFO - Inserted 194 products into bronze
2026-04-13 14:31:34 - bronze_to_silver  - INFO - Inserted 194 products into silver.products
2026-04-13 14:31:34 - bronze_to_silver  - INFO - Inserted 364 tags into silver.product_tags
2026-04-13 14:31:34 - bronze_to_silver  - INFO - Inserted 474 images into silver.product_images
2026-04-13 14:31:34 - bronze_to_silver  - INFO - Inserted 582 reviews into silver.product_reviews

---

Infrastructure & Containerization

The system is fully containerized using Docker to ensure environment parity and automated deployment. The architecture consists of a coordinated multi-container setup:

Database (Postgres 15): * Automated Schema Management: Uses docker-entrypoint-initdb.d to execute SQL migrations automatically on startup (Bronze, Silver, Gold layers).

Healthchecks: Implemented pg_isready to ensure the database engine is fully operational before any data ingestion starts.

Persistence: Managed via Docker volumes (pgdata) to ensure data integrity across container restarts.

Data Pipeline (Python 3.10-slim):

Service Dependency: Configured with condition: service_healthy to prevent execution before the database is ready.

Environment-Driven: All configurations (DB credentials, API endpoints) are managed via environment variables for security and scalability.

Commands:

docker-compose up --build : Start system & migrations.

docker-compose down -v : Full cleanup.

Tech Stack

Tool	Purpose
Python 3.10+	Pipeline logic
PostgreSQL 14+	Data warehouse
psycopg2	PostgreSQL driver
JSONB	Raw payload storage
SQL migrations	Schema versioning

---

Data Source

DummyJSON Products API — 194 products across multiple categories including beauty, groceries, fragrances, and furniture.

---

License

MIT