my-pipeline/
+-- pyproject.toml          # Dependencies, Build-Config, Tool-Settings
+-- src/
|   +-- my_pipeline/
|       +-- __init__.py
|       +-- extract.py      # Source-Connectoren
|       +-- transform.py    # Business-Logic
|       +-- load.py         # Sink-Writer
|       +-- models.py       # Pydantic-Schemas
+-- tests/
|   +-- test_extract.py
|   +-- test_transform.py
+-- .env.example
+-- README.md

# uv installieren
curl -LsSf https://astral.sh/uv/install.sh | sh

# Neues Projekt initialisieren
uv init my-pipeline
cd my-pipeline

# Dependencies hinzufügen
uv add pandas sqlalchemy psycopg2-binary pydantic
uv add --dev pytest ruff mypy

# Skripte in managed Environment ausführen
uv run python src/my_pipeline/extract.py

# Lock und sync
uv lock
uv sync

[project]
name = "my-pipeline"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "pandas>=2.2",
    "sqlalchemy>=2.0",
    "psycopg2-binary>=2.9",
    "pydantic>=2.0",
]

[tool.ruff]
target-version = "py311"
select = ["E", "F", "I", "UP"]

# models.py
from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings

class DatabaseConfig(BaseSettings):
    host: str = Field(..., alias="DB_HOST")
    port: int = Field(5432, alias="DB_PORT")
    database: str = Field(..., alias="DB_NAME")
    user: str = Field(..., alias="DB_USER")
    password: SecretStr = Field(..., alias="DB_PASSWORD")

    @property
    def connection_string(self) -> str:
        return (
            f"postgresql://{self.user}:"
            f"{self.password.get_secret_value()}"
            f"@{self.host}:{self.port}/{self.database}"
        )

    model_config = {"env_file": ".env"}


class PipelineConfig(BaseSettings):
    db: DatabaseConfig = DatabaseConfig()
    batch_size: int = Field(10_000, alias="BATCH_SIZE")
    max_retries: int = Field(3, alias="MAX_RETRIES")
    source_api_url: str = Field(..., alias="SOURCE_API_URL")

# extract.py
from my_pipeline.models import PipelineConfig

config = PipelineConfig()
# Wenn DB_HOST oder SOURCE_API_URL nicht gesetzt sind,
# wirft Pydantic einen klaren ValidationError beim Startup

# extract.py
import httpx
import time
import logging
from typing import Generator

logger = logging.getLogger(__name__)

def fetch_paginated(
    base_url: str,
    headers: dict,
    page_size: int = 100,
    max_retries: int = 3,
) -> Generator[list[dict], None, None]:
    offset = 0

    with httpx.Client(timeout=30.0) as client:
        while True:
            params = {"limit": page_size, "offset": offset}

            for attempt in range(1, max_retries + 1):
                try:
                    resp = client.get(
                        base_url,
                        headers=headers,
                        params=params,
                    )
                    resp.raise_for_status()
                    break
                except httpx.HTTPStatusError as e:
                    if e.response.status_code == 429:
                        wait = int(
                            e.response.headers.get("Retry-After", 2 ** attempt)
                        )
                        logger.warning(f"Rate limited. Warte {wait}s.")
                        time.sleep(wait)
                    elif attempt == max_retries:
                        raise
                    else:
                        time.sleep(2 ** attempt)
                except httpx.RequestError:
                    if attempt == max_retries:
                        raise
                    time.sleep(2 ** attempt)

            data = resp.json()
            records = data.get("results", [])

            if not records:
                break

            yield records
            offset += page_size

            logger.info(f"Bisher {offset} Records geladen.")

# load.py -- PostgreSQL-Dialekt
from sqlalchemy import create_engine, text
from typing import Iterator
import logging

logger = logging.getLogger(__name__)

def batch_upsert(
    connection_string: str,
    table: str,
    records: Iterator[list[dict]],
    conflict_key: str = "id",
    batch_size: int = 5_000,
) -> int:
    engine = create_engine(connection_string)
    total = 0

    with engine.begin() as conn:
        for batch in records:
            if not batch:
                continue

            columns = list(batch[0].keys())
            placeholders = ", ".join(f":{col}" for col in columns)
            update_clause = ", ".join(
                f"{col} = EXCLUDED.{col}"
                for col in columns
                if col != conflict_key
            )

            # PostgreSQL INSERT ... ON CONFLICT (upsert)
            stmt = text(
                f"INSERT INTO {table} ({', '.join(columns)}) "
                f"VALUES ({placeholders}) "
                f"ON CONFLICT ({conflict_key}) "
                f"DO UPDATE SET {update_clause}"
            )

            conn.execute(stmt, batch)
            total += len(batch)
            logger.info(f"{total} Records nach {table} upserted.")

    return total

# Pandas -- vertraut, gut für schnelle Exploration
# Gut für: <500 MB, gemischte Operationen, breites Ökosystem
import pandas as pd

df = pd.read_parquet("events.parquet")
result = (
    df.groupby("event_type")
    .agg(count=("event_id", "count"), avg_value=("value", "mean"))
    .reset_index()
    .sort_values("count", ascending=False)
)

# Polars -- gleiche Logic, 5-10x schneller auf mittleren Datasets
# Gut für: 500 MB-10 GB, aggregation-lastig, Lazy Evaluation
import polars as pl

result = (
    pl.scan_parquet("events.parquet")  # lazy -- noch keine Daten geladen
    .group_by("event_type")
    .agg(
        pl.col("event_id").count().alias("count"),
        pl.col("value").mean().alias("avg_value"),
    )
    .sort("count", descending=True)
    .collect()  # Ausführung hier
)

# transform.py -- reine Funktionen, kein I/O
import polars as pl

def clean_events(df: pl.DataFrame) -> pl.DataFrame:
    return (
        df.filter(pl.col("event_id").is_not_null())
        .unique(subset=["event_id"])
        .with_columns(
            pl.col("timestamp").str.to_datetime().alias("event_ts"),
            pl.col("value").fill_null(0).alias("value_clean"),
        )
    )

def flag_anomalies(
    df: pl.DataFrame,
    threshold: float = 3.0,
) -> pl.DataFrame:
    stats = df.select(
        pl.col("value_clean").mean().alias("mu"),
        pl.col("value_clean").std().alias("sigma"),
    ).row(0)
    mu, sigma = stats

    return df.with_columns(
        (
            ((pl.col("value_clean") - mu).abs() > threshold * sigma)
        ).alias("is_anomaly")
    )

# tests/test_transform.py
import polars as pl
from my_pipeline.transform import clean_events, flag_anomalies

def test_clean_events_removes_nulls():
    df = pl.DataFrame({
        "event_id": ["a", None, "b", "a"],
        "timestamp": ["2026-01-01", "2026-01-02", "2026-01-03", "2026-01-01"],
        "value": [10.0, 20.0, None, 10.0],
    })
    result = clean_events(df)

    assert result.shape[0] == 2  # null entfernt, Duplikat entfernt
    assert result.filter(pl.col("event_id").is_null()).shape[0] == 0

def test_flag_anomalies_catches_outliers():
    df = pl.DataFrame({
        "value_clean": [10.0, 11.0, 9.0, 10.5, 100.0],
    })
    result = flag_anomalies(df, threshold=2.0)

    anomalies = result.filter(pl.col("is_anomaly"))
    assert anomalies.shape[0] == 1
    assert anomalies["value_clean"][0] == 100.0

# Lädt 50 GB in RAM und crasht
df = pd.read_parquet("s3://bucket/huge-table/")

# Partition für Partition verarbeiten
import pyarrow.parquet as pq

dataset = pq.ParquetDataset("s3://bucket/huge-table/")
for batch in dataset.to_batches(batch_size=100_000):
    chunk = pl.from_arrow(batch)
    process(chunk)

query = f"SELECT * FROM users WHERE name = '{user_input}'"

# SQLAlchemy -- PostgreSQL-Dialekt
from sqlalchemy import text
query = text("SELECT * FROM users WHERE name = :name")
result = conn.execute(query, {"name": user_input})

print(f"Processed {count} records")  # In Container-Logs verloren

import logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)
logger.info("Processed %d records", count)

# Ohne Types -- was returnt das? Was ist 'config'?
def extract(config, start_date):
    ...

# Mit Types -- klarer Vertrag
def extract(
    config: PipelineConfig,
    start_date: datetime,
) -> Generator[list[dict], None, None]:
    ...

import asyncio
import httpx

async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        return [
            r.json() for r in responses
            if isinstance(r, httpx.Response) and r.status_code == 200
        ]

# Verwendung
results = asyncio.run(fetch_all(api_endpoints))

# main.py -- Orchestrierungs-Entrypoint
import logging
from my_pipeline.models import PipelineConfig
from my_pipeline.extract import fetch_paginated
from my_pipeline.transform import clean_events, flag_anomalies
from my_pipeline.load import batch_upsert
import polars as pl

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def run():
    config = PipelineConfig()  # validiert Env-Vars oder crasht

    # Extract: paginierter API-Fetch (Generator)
    raw_pages = fetch_paginated(
        base_url=config.source_api_url,
        headers={"Authorization": f"Bearer {config.api_token}"},
        page_size=config.batch_size,
    )

    # Transform: jede Seite verarbeiten, gereinigte Batches yielden
    def transform_stream(pages):
        for page in pages:
            df = pl.DataFrame(page)
            df = clean_events(df)
            df = flag_anomalies(df)
            yield df.to_dicts()

    # Load: Upsert in PostgreSQL
    total = batch_upsert(
        connection_string=config.db.connection_string,
        table="events",
        records=transform_stream(raw_pages),
        conflict_key="event_id",
    )

    logger.info(f"Pipeline fertig. {total} Records upserted.")

if __name__ == "__main__":
    run()