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 csvimport osfrom datetime import datefrom 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)