JSON Export

Provides patterns for JSON output generation in ETL pipelines.

Basic JSON Export

import json
from pathlib import Path
from typing import Any
from datetime import datetime, date
from decimal import Decimal

def json_serializer(obj: Any) -> Any:
    """Custom JSON serializer for special types."""
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    elif isinstance(obj, Decimal):
        return float(obj)
    elif hasattr(obj, '__dict__'):
        return obj.__dict__
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")

def write_json(filepath: str, data: Any, indent: int = 2):
    """Write data to JSON file with proper formatting."""
    Path(filepath).parent.mkdir(parents=True, exist_ok=True)
    with open(filepath, 'w', encoding='utf-8') as f:
        json.dump(data, f, indent=indent, default=json_serializer, ensure_ascii=False)

def to_json_string(data: Any, indent: int = None) -> str:
    """Convert data to JSON string."""
    return json.dumps(data, indent=indent, default=json_serializer, ensure_ascii=False)

Structured Reports

from datetime import datetime
from typing import List, Dict, Any

def create_report(
    title: str,
    data: Any,
    metadata: Dict[str, Any] = None
) -> Dict[str, Any]:
    """Create a structured report with metadata."""
    return {
        'report_title': title,
        'generated_at': datetime.now().isoformat(),
        'metadata': metadata or {},
        'data': data
    }

def create_summary_report(
    summaries: Dict[str, Any],
    details: List[Dict] = None
) -> Dict[str, Any]:
    """Create a summary report with optional details."""
    return {
        'generated_at': datetime.now().isoformat(),
        'summary': summaries,
        'record_count': len(details) if details else 0,
        'details': details
    }