|
""" |
|
API to Sheets Demo - Mock JSON to CSV conversion script. |
|
|
|
This module provides functionality to load JSON data, validate and process it, |
|
and export the results to CSV format. Designed as a demonstration of API |
|
integration patterns with production-ready error handling and validation. |
|
""" |
|
|
|
import json |
|
import csv |
|
from datetime import datetime |
|
|
|
|
|
class SheetSync: |
|
""" |
|
Handles loading, processing, and exporting data from JSON to CSV format. |
|
|
|
Provides data validation, error handling, and transformation capabilities |
|
for converting API-like JSON responses into structured CSV output. |
|
""" |
|
|
|
REQUIRED_KEYS = ['id', 'name', 'score'] |
|
|
|
def load_data(self, json_path): |
|
""" |
|
Load and process JSON data from a file. |
|
|
|
Args: |
|
json_path (str): Path to the JSON file to load. |
|
|
|
Returns: |
|
list: Processed list of dictionaries with lowercase keys and |
|
processed timestamps. Returns empty list on error. |
|
|
|
Raises: |
|
FileNotFoundError: If the JSON file doesn't exist. |
|
PermissionError: If file cannot be read due to permissions. |
|
json.JSONDecodeError: If JSON is malformed. |
|
""" |
|
try: |
|
with open(json_path, 'r', encoding='utf-8') as f: |
|
data = json.load(f) |
|
except FileNotFoundError: |
|
print(f"Error: File '{json_path}' not found.") |
|
return [] |
|
except PermissionError: |
|
print(f"Error: Permission denied reading '{json_path}'.") |
|
return [] |
|
except json.JSONDecodeError as e: |
|
print(f"Error: Invalid JSON in '{json_path}': {e}") |
|
return [] |
|
|
|
if not isinstance(data, list): |
|
print(f"Error: Expected JSON array, got {type(data).__name__}.") |
|
return [] |
|
|
|
processed = [] |
|
skipped = 0 |
|
|
|
for item in data: |
|
if not isinstance(item, dict): |
|
print(f"Warning: Skipping non-dict item: {item}") |
|
skipped += 1 |
|
continue |
|
|
|
required_keys_lower = [k.lower() for k in self.REQUIRED_KEYS] |
|
item_keys_lower = [k.lower() for k in item.keys()] |
|
|
|
if not all(key in item_keys_lower for key in required_keys_lower): |
|
print(f"Warning: Skipping item missing required keys: {item}") |
|
skipped += 1 |
|
continue |
|
|
|
processed_item = {k.lower(): v for k, v in item.items()} |
|
processed_item['processed_timestamp'] = datetime.now().isoformat() |
|
processed.append(processed_item) |
|
|
|
if skipped > 0: |
|
print(f"Skipped {skipped} invalid record(s).") |
|
|
|
print(f"Loaded {len(processed)} record(s).") |
|
return processed |
|
|
|
def write_csv(self, data, output_path): |
|
""" |
|
Write processed data to a CSV file. |
|
|
|
Args: |
|
data (list): List of dictionaries to write to CSV. |
|
output_path (str): Path where the CSV file will be written. |
|
|
|
Returns: |
|
bool: True if successful, False otherwise. |
|
|
|
Raises: |
|
PermissionError: If file cannot be written due to permissions. |
|
""" |
|
if not data: |
|
print("Warning: No data to write. CSV file not created.") |
|
return False |
|
|
|
try: |
|
fieldnames = list(data[0].keys()) |
|
|
|
with open(output_path, 'w', newline='', encoding='utf-8') as f: |
|
writer = csv.DictWriter(f, fieldnames=fieldnames) |
|
writer.writeheader() |
|
writer.writerows(data) |
|
|
|
print(f"Wrote CSV to {output_path}") |
|
return True |
|
except PermissionError: |
|
print(f"Error: Permission denied writing to '{output_path}'.") |
|
return False |
|
except Exception as e: |
|
print(f"Error writing CSV: {e}") |
|
return False |
|
|
|
|
|
if __name__ == '__main__': |
|
sync = SheetSync() |
|
data = sync.load_data('data.json') |
|
sync.write_csv(data, 'output.csv') |
|
|