Before You Begin

  • Python 3.10+ (type hints with | syntax used)
  • pip install requests pandas pydantic aiohttp aiosqlite
  • Basic familiarity with requests and pandas (level of existing beginner tutorial)
  • No prior async experience required – we’ll build it step by step

Problem: Messy Fruit Market Data

You have access to three different APIs that report fruit prices, but each returns data in a different format. Some fields are missing, prices use inconsistent units (per kg vs per lb), and timestamps are in varying timezones. You need a single, clean dataset to run analytics (e.g., “which fruit is most volatile this week?”).

Manual cleaning doesn’t scale. You need a pipeline that:

  • Fetches data concurrently (because APIs are slow)
  • Validates and normalises every record
  • Handles failures gracefully without losing the whole batch
  • Stores the result for later analysis

Solution: Build a Pipeline with Four Steps

We’ll construct a reusable FruitPipeline class that orchestrates: Extract, Validate, Clean, Load. Each step is a distinct module, making the system testable and maintainable.

1. Extract – Async Scraping with Retries

Synchronous scraping (requests) blocks the CPU while waiting for I/O. Production codebases on GitHub (e.g., Scrapy, aiohttp examples) use asyncio to fire multiple requests concurrently.

import asyncio
import aiohttp
from typing import Dict, Any

FRUIT_APIS = [
    "https://fruit-market-1.example.com/prices",
    "https://fruit-market-2.example.com/prices",
    "https://fruit-market-3.example.com/prices",
]

async def fetch(session: aiohttp.ClientSession, url: str) -> list[Dict[str, Any]]:
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            resp.raise_for_status()
            data = await resp.json()
            return data  # assume API returns a list of fruit objects
    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
        # Common pitfall on StackOverflow: silent failure. We log and return empty.
        print(f"Failed to fetch {url}: {e}")
        return []

async def extract_all() -> list[Dict[str, Any]]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in FRUIT_APIS]
        results = await asyncio.gather(*tasks)
    # flatten list of lists
    return [item for sublist in results for item in sublist]

> Engineering blog tip: Netflix’s engineering blog recommends using asyncio.gather(return_exceptions=True) when you want partial results. Here we keep it simple: failures return empty lists.

2. Validate – Pydantic Models for Data Integrity

Raw JSON can have missing keys, wrong types, or unexpected fields. Pydantic (used in FastAPI, LangChain) provides strict validation with clear error messages.

Define a model for a fruit record:

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Literal

class FruitRecord(BaseModel):
    fruit: str
    price_per_kg: float = Field(alias="price_kg", ge=0)  # API field name may differ
    unit: Literal["kg", "lb"]
    timestamp: datetime
    market: str

    @classmethod
    def from_raw(cls, raw: dict) -> "FruitRecord | None":
        try:
            # map common variations
            if "price" in raw and "unit" not in raw:
                raw["unit"] = "kg"  # default assumption
            return cls.model_validate(raw, strict=False)
        except Exception as e:
            print(f"Validation error: {e} for record {raw.get('fruit', '?')}")
            return None

from_raw is a common pattern (seen in many OSS projects) to handle dirty data without crashing the pipeline.

3. Clean – Normalise and Deduplicate with Pandas

Even validated data often needs conversions (e.g., lb → kg) and duplicate removal.

import pandas as pd
from typing import List

def clean(records: List[FruitRecord]) -> pd.DataFrame:
    df = pd.DataFrame([r.model_dump() for r in records])
    # Convert price per lb to price per kg (1 lb ≈ 0.4536 kg)
    lb_idx = df["unit"] == "lb"
    df.loc[lb_idx, "price_per_kg"] = df.loc[lb_idx, "price_per_kg"] / 0.4536
    df["unit"] = "kg"  # now all in kg
    # Drop exact duplicates (same fruit, same market, same timestamp)
    df.drop_duplicates(subset=["fruit", "market", "timestamp"], inplace=True)
    # Resample or aggregate? Not yet – we keep raw for storage.
    return df

4. Load – Store in SQLite for Persistence

Use aiosqlite to stay in the async context (optional, but keeps the pipeline fully async).

import aiosqlite

async def load_to_db(df: pd.DataFrame, db_path: str = "fruits.db"):
    async with aiosqlite.connect(db_path) as db:
        # Create table if not exists
        await db.execute("""
            CREATE TABLE IF NOT EXISTS fruit_prices (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                fruit TEXT,
                price_per_kg REAL,
                timestamp TEXT,
                market TEXT
            )
        """)
        # Insert using executemany
        records = df.to_dict(orient="records")
        await db.executemany(
            "INSERT INTO fruit_prices (fruit, price_per_kg, timestamp, market) VALUES (?,?,?,?)",
            [(r["fruit"], r["price_per_kg"], r["timestamp"].isoformat(), r["market"]) for r in records]
        )
        await db.commit()

5. Orchestrate the Pipeline

Tie everything together with error handling and logging:

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class FruitPipeline:
    def __init__(self, db_path: str = "fruits.db"):
        self.db_path = db_path

    async def run(self):
        logger.info("Starting fruit pipeline")
        # Extract
        raw_data = await extract_all()
        logger.info(f"Extracted {len(raw_data)} raw records")
        # Validate
        valid_records = []
        for raw in raw_data:
            fruit = FruitRecord.from_raw(raw)
            if fruit:
                valid_records.append(fruit)
        logger.info(f"Validated {len(valid_records)} records")
        # Clean
        df = clean(valid_records)
        logger.info(f"Cleaned dataframe: {df.shape}")
        # Load
        await load_to_db(df, self.db_path)
        logger.info("Pipeline complete")

# Entry point
if __name__ == "__main__":
    pipeline = FruitPipeline()
    asyncio.run(pipeline.run())

Common Issues & Solutions

  • SSL certificate errors when scraping: Add ssl=False to ClientSession (not recommended production) or create a custom SSL context. Official aiohttp docs explain the secure way.
  • Pydantic modelvalidate fails on extra fields: Use modelconfig = {"extra": "ignore"} in the model class.
  • aiosqlite not installed: The library is popular on GitHub but not built-in; install with pip install aiosqlite. Many developers on StackOverflow forget this.
  • Duplicates in database: Add a unique constraint UNIQUE(fruit, market, timestamp) to the SQLite table, and use INSERT OR IGNORE.
Sponsored Deal

Extending the Pipeline

  • Add retry logic with exponential backoff using tenacity library (seen in many AWS Lambda functions).
  • Send alerts via Slack if validation error rate exceeds 10%.
  • Use Apache Arrow (PyArrow) instead of Pandas for memory efficiency when data volumes grow.

This intermediate pipeline gives you a solid foundation to build on – you now own a maintainable, resilient data processing system in Python.