63 lines
jobs/export_job.py
Exports filtered user records to CSV given a date range.
# User account export job for the analytics platform.
import csv
import os
from datetime import date
from typing import List
 
# Type alias for a user record dict.
User = dict  # keys: user_id (str), email (str), created_at (date), is_active (bool)
 
 
# Parameters: date_from, date_to — start and end dates for the export window.
# Returns: None. Raises ValueError when date_from is after date_to.
def _validate_date_range(date_from: date, date_to: date) -> None:
    if date_from > date_to:
        raise ValueError(
            f"date_from ({date_from}) must not be after date_to ({date_to})"
        )
 
 
def export_users(
    users: List[User],
    path: str,
    date_from: date,
    date_to: date,
) -> int:
    """Export active users created within [date_from, date_to] to a CSV file.
 
    Both date_from and date_to are inclusive. Only active users are exported.
 
    Parameters
    ----------
    users : list of User
        Full list of user records to filter.
    path : str
        Destination file path; parent directories are created if needed.
    date_from : date
        Earliest creation date to include (inclusive).
    date_to : date
        Latest creation date to include (inclusive).
 
    Returns
    -------
    int
        Number of rows written, excluding the header.
    """
    _validate_date_range(date_from, date_to)
 
    filtered = [
        u for u in users
        if date_from <= u["created_at"] < date_to
    ]
 
    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
    with open(path, "w", newline="") as fh:
        writer = csv.DictWriter(
            fh,
            fieldnames=["user_id", "email", "created_at"],
            extrasaction="ignore",
        )
        writer.writeheader()
        writer.writerows(filtered)
 
    return len(filtered)