Most ETL tutorials show a perfect, frictionless flow. The reality? My pipeline turned into a festival of KeyError's, outdated schemas, and API payload typos.
In this article, I will walk you through a complete Extrac -> transform -> Load pipeline using Python, Docker, and PostgreSqul. But more importantly, I'll share the real errors I ran into anh how I debugged them. Because calmly reading tracebacks is the only realible way to fix data pipelines when they inevitably fail.
What we'ew building
GitHub API -> extract.py -> transform.py -> load.py -> PostgreSQL (Docker)
- Extract: Paginated fetch of issues from a public repository via the GitHub REST API.
- Transform: Normalizes each issue into a flat record and calculates the hours it took to close.
- Load: Creates the database table if it doesn't exist ans upserts by ID, so re-running the pipeline updates rows instead of duplicating them.
The Stack
- Python 3.14
-
psycopg(v3) to connect to Postgres -
python-dotenvfor configuration management - PostgreSQL running in Docker Compose
Windows + Python 3.14 Note: If you are on Windows, use
psycopg[binary]in your requirements instead ofpsycopg2-binary. The latter often fails to compile cleanly on this combination.
1. Proyect Structure
.
├── extract.py
├── transform.py
├── load.py
├── main.py
├── docker-compose.yml
├── requirements.txt
└── .env.example
2. Spinning up Postgres with Docker
First let's get our database running. Using Docker Compose keeps our local enviroment clean.
services:
postgres:
image: postgres:16-alpine
environment:
POSTGRES_USER: etl_user
POSTGRES_PASSWORD: etl_password
POSTGRES_DB: github_analytics
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
volumes:
pgdata:
Start the database with:
docker compose up -d
3. Configuration via .env
Create your .env file with the necessary credentials:
GITHUB_TOKEN=your_github_personal_access_token
GITHUB_REPO=microsoft/vscode
DB_HOST=localhost
DB_PORT=5432
DB_NAME=github_analytics
DB_USER=etl_user
DB_PASSWORD=etl_password
The Common Trap: KeyError: GitHUB_REPO.env
**The Lesson:** __If you forget to createfrom.env.example, or if you forget callload_dotenv()inmain.py`, the script blows up immediately. Always chechks your enviroment variables first when a pipeline fails at startup.__
4. Extract: Pulling the Issues
We use the request library to handle pagination from the GitHub API.
`python
extract.py
import os
import logging
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
logger = logging.getLogger(name)
GITHUB_API = "https://api.github.com"
def _headers():
token = os.environ.get("GITHUB_TOKEN")
headers = {"Accept": "application/vnd.github+json"}
if token:
headers["Authorization"] = f"Bearer {token}"
return headers
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def _get(url, params=None):
response = requests.get(url, headers=_headers(), params=params, timeout=10)
response.raise_for_status()
return response
def fetch_issues(repo: str, max_pages: int = 5):
page = 1
while page <= max_pages:
logger.info("Fetching %s issues page %d", repo, page)
resp = _get(
f"{GITHUB_API}/repos/{repo}/issues",
params={"state": "all", "per_page": 100, "page": page},
)
batch = resp.json()
if not batch:
break
for item in batch:
yield item
page += 1
`
5. Transform: Normalizing and Computing Metrics
This is what a standard GitHub API response looks like for an issue
json
{
"id": 1234567,
"number": 42,
"title": "Bug in the auth module",
"state": "closed",
"created_at": "2024-01-01T10:00:00Z",
"closed_at": "2024-01-02T12:00:00Z",
"user": {
"login": "octocat"
}
}
Here is our transformation logic:
`python
transform.py
from datetime import datetime
def normalize_issue(raw: dict) -> dict:
return {
"id": raw["id"],
"number": raw["number"],
"title": raw["title"],
"state": raw["state"],
"is_pull_request": "pull_request" in raw,
"created_at": raw["created_at"],
"closed_at": raw.get("closed_at"),
"comments": raw["comments"],
"author": raw["user"]["login"] if raw.get("user") else None,
}
def time_to_close_hours(row: dict):
if not row["closed_at"]:
return None
created = datetime.fromisoformat(row["created_at"].replace("Z", "+00:00"))
closed = datetime.fromisoformat(row["closed_at"].replace("Z", "+00:00"))
return round((closed - created).total_seconds() / 3600, 2)
`
The Typo Cascade: KeyError: "author"
** The Lesson:** I wrote raw["author"] assuming the fiels name, but GitHub uses raw["user]["login"]. Field names are a contract. Never guess them from memory. ALways validate your transformation logic against the actual JSON payload or API documentation
6. Load: Creating the Schema and Upserting
`python
load.py
import logging
import psycopg
logger = logging.getLogger(name)
CREATE_TABLE = """
CREATE TABLE IF NOT EXISTS issues (
id BIGINT PRIMARY KEY,
number INT,
title TEXT,
state TEXT,
is_pull_request BOOLEAN,
created_at TIMESTAMPTZ,
closed_at TIMESTAMPTZ,
comments INT,
author TEXT,
time_to_close_hours NUMERIC
);
"""
UPSERT = """
INSERT INTO issues (id, number, title, state, is_pull_request, created_at, closed_at, comments, author, time_to_close_hours)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (id) DO UPDATE SET
state = EXCLUDED.state,
closed_at = EXCLUDED.closed_at,
comments = EXCLUDED.comments,
time_to_close_hours = EXCLUDED.time_to_close_hours;
"""
def get_connection(db_config: dict):
return psycopg.connect(**db_config)
def ensure_schema(conn):
with conn.cursor() as cur:
cur.execute(CREATE_TABLE)
conn.commit()
def load_rows(conn, rows: list[dict]):
if not rows:
logger.info("No rows to load.")
return
values = [
(r["id"], r["number"], r["title"], r["state"], r["is_pull_request"],
r["created_at"], r["closed_at"], r["comments"], r["author"], r["time_to_close_hours"])
for r in rows
]
with conn.cursor() as cur:
cur.executemany(UPSERT, values)
conn.commit()
logger.info("Loaded/updated %d rows.", len(rows))
`
The Schema Error:
psycopg.errors.UndefinedColumn: column "closed_at" of relation "issues" does not exist
The Lesson: __CREATE TABLE IF NOT EXISTS does exactly what it says. If the table already existis, it does nothing-even if you added a new colum (closet_at) to your SQL string. It is a migration tool.
TO fix this during local development, drop the table using Docker Compose so the script can recreate it with the new schema:
`bash
docker compose exec postgres psql -U etl_user -d github_analytics -c "DROP TABLE IF EXISTS issues;"
`
7. Querying the Data
Once your main.py runs successfully, you can query your data directly through the container:
`bash
docker compose exec postgres psql -U etl_user -d github_analytics
`
What I took away from this project
None of the errors I ran into were mathematically complex -they were typos, schema mismatches, and misremembered fields. What mattered was the order in which I tackled them: one traceback at a time, always reading fro the bottom up to find the root cause.
If you're building your own ETL, my advice is simple: run the pipeline often, in small steps, and let each error lead you to the next one. It's the most reliable way to end up with a robust architecture.
Next Steps: Taking it to the Cloud
Running this locally in DOcker Compose is perfect for development and debugging, but the natural next step for any ETL pipeline is moving it to a production environment.
My next goal for this project is to migrate this local stack into the cloud -likely containerizing the Python scripts as a CronJob and hosting the PostgreDQL database on a fast developer-friendly cloud provider (like Civo's Kubernetes or Compute instances) to fully automate the extractionlayer.
Full code: The complete repository is on GitHub: github.com/WhoIsMarce/github-analytics-etl-pipeline
Over to you: What's the most ridiculous typo or silent error that ever broke your data pipeline? Drop it in the comments so I feel less alone!
Top comments (0)