First Post

WITH daily_activity AS (
    SELECT
        u.id AS user_id,
        DATE_TRUNC('week', u.created_at) AS cohort_week,
        d.activity_date,
        COUNT(DISTINCT d.session_id) AS sessions
    FROM users u
    JOIN daily_events d ON u.id = d.user_id
    WHERE d.activity_date >= CURRENT_DATE - INTERVAL '90 days'
    GROUP BY u.id, cohort_week, d.activity_date
),
rolling_active AS (
    SELECT
        cohort_week,
        activity_date,
        COUNT(DISTINCT user_id) AS active_users,
        SUM(sessions) AS total_sessions
    FROM daily_activity
    GROUP BY cohort_week, activity_date
)
SELECT
    cohort_week,
    activity_date,
    active_users,
    total_sessions,
    ROUND(total_sessions::NUMERIC / active_users, 2) AS avg_sessions_per_user
FROM rolling_active
ORDER BY cohort_week, activity_date;

import logging
from dataclasses import dataclass
from datetime import datetime

import psycopg2
from psycopg2.extras import execute_values

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


@dataclass
class PipelineConfig:
    source_dsn: str
    target_dsn: str
    batch_size: int = 5000


def extract(config: PipelineConfig, since: datetime) -> list[dict]:
    """Pull raw event records from the source database."""
    query = """
        SELECT user_id, event_type, payload, created_at
        FROM raw_events
        WHERE created_at >= %s
        ORDER BY created_at
    """
    with psycopg2.connect(config.source_dsn) as conn:
        with conn.cursor() as cur:
            cur.execute(query, (since,))
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]


def transform(records: list[dict]) -> list[tuple]:
    """Clean and reshape records for the analytics schema."""
    transformed = []
    for r in records:
        event_date = r["created_at"].date()
        category = classify_event(r["event_type"])
        transformed.append((
            r["user_id"],
            event_date,
            r["event_type"],
            category,
            r["payload"],
        ))
    return transformed


def classify_event(event_type: str) -> str:
    mapping = {
        "page_view": "engagement",
        "signup": "acquisition",
        "purchase": "monetization",
        "referral": "growth",
    }
    return mapping.get(event_type, "other")


def load(config: PipelineConfig, rows: list[tuple]) -> int:
    """Bulk-insert transformed rows into the target table."""
    insert_sql = """
        INSERT INTO analytics_events
            (user_id, event_date, event_type, category, payload)
        VALUES %s
        ON CONFLICT (user_id, event_date, event_type) DO NOTHING
    """
    with psycopg2.connect(config.target_dsn) as conn:
        with conn.cursor() as cur:
            for i in range(0, len(rows), config.batch_size):
                batch = rows[i : i + config.batch_size]
                execute_values(cur, insert_sql, batch)
        conn.commit()
    return len(rows)


if __name__ == "__main__":
    cfg = PipelineConfig(
        source_dsn="postgresql://src_user:pass@localhost:5432/source_db",
        target_dsn="postgresql://tgt_user:pass@localhost:5432/target_db",
    )
    since = datetime(2026, 1, 1)
    logger.info("Starting extraction since %s", since)

    raw = extract(cfg, since)
    logger.info("Extracted %d records", len(raw))

    cleaned = transform(raw)
    logger.info("Transformed %d records", len(cleaned))

    count = load(cfg, cleaned)
    logger.info("Loaded %d records into target", count)