kerbemkator/african-news-pipeline

Async ETL pipeline ingesting 53 African news RSS feeds across 26 countries — Pydantic validation, PostgreSQL storage, Celery scheduling # African News Intelligence Pipeline ### Phase 2 — AI Engineering Roadmap: Python & Data Engineering > African news is fragmented across dozens of portals. This pipeline ingests RSS feeds > from 53 African news sources across 26 countries, normalizes, deduplicates, and stores > structured data in PostgreSQL with scheduled refresh via Celery. --- ## Architecture ``` RSS Sources (53) → Async Scraper (aiohttp) → Pydantic Validation ↓ PostgreSQL ← Bulk Upsert (ON CONFLICT) ← Dedup (SHA-256 hash) ← Normalizer ↑ Celery Beat (every 30 min) ``` ## Stack - **aiohttp** — concurrent async HTTP requests (10 concurrent, semaphore-bounded) - **feedparser** — RSS/Atom parsing - **Pydantic v2** — schema validation, rejects malformed entries - **PostgreSQL + asyncpg** — async storage with `ON CONFLICT DO NOTHING` dedup - **Redis** — fast-path duplicate cache - **Celery + Celery Beat** — scheduled pipeline runs every 30 minutes ## Key Results - 53 sources across 26 African countries - Deduplication via SHA-256 content hashing (title + source, case-insensitive) - Database-level conflict handling — race-safe across concurrent runs - Full async pipeline — non-blocking I/O throughout ## Project Structure ``` african-news-pipeline/ ├── src/ │ ├── config.py # Environment settings │ ├── models.py # Pydantic schemas + SQLAlchemy ORM │ ├── database.py # Async PostgreSQL + upsert logic │ ├── cache.py # Redis dedup cache │ ├── tasks.py # Celery scheduled tasks │ ├── scrapers/ │ │ ├── sources.py # 53 RSS feed definitions │ │ └── rss_scraper.py # Async fetch + parse │ └── pipeline/ │ ├── normalizer.py # Hash + normalize │ └── etl.py # Full orchestration ├── tests/test_pipeline.py ├── docker-compose.yml # PostgreSQL + Redis └── requirements.txt ``` ## Setup ```bash git clone github.com …