import argparse import csv import json import logging import sqlite3 import sys import time from datetime import datetime from typing import Any, Dict, List, Optional, Tuple import httpx # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def read_csv(file_path: str) -> List[Dict[str, Any]]: """ Read data from a CSV file. Args: file_path: Path to the CSV file. Returns: List of dictionaries representing the rows. Raises: Exception: If file cannot be read. """ try: with open(file_path, 'r', newline='', encoding='utf-8') as f: reader = csv.DictReader(f) return [dict(row) for row in reader] except Exception as e: logger.error(f"Failed to read CSV file: {e}") raise def read_json(file_path: str) -> List[Dict[str, Any]]: """ Read data from a JSON file. Args: file_path: Path to the JSON file. Returns: List of dictionaries representing the JSON data. Raises: Exception: If file cannot be read. """ try: with open(file_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f"Failed to read JSON file: {e}") raise def read_api(url: str) -> List[Dict[str, Any]]: """ Read data from an API. Args: url: URL of the API endpoint. Returns: List of dictionaries representing the API response. Raises: Exception: If API call fails. """ try: with httpx.Client() as client: response = client.get(url) response.raise_for_status() return response.json() except httpx.HTTPError as e: logger.error(f"API request failed: {e}") raise def apply_transformations(data: List[Dict[str, Any]], transformations: Dict[str, str]) -> List[Dict[str, Any]]: """ Apply transformation rules to data. Args: data: List of dictionaries to transform. transformations: Mapping of field names to transformation rules. Returns: Transformed data. """ transformed = [] for row in data: transformed_row = {} for key, rule in transformations.items(): if key in row: if rule == "upper": transformed_row[key] = row[key].upper() elif rule == "lower": transformed_row[key] = row[key].lower() elif rule == "strip": transformed_row[key] = row[key].strip() elif rule.startswith("format:"): format_str = rule[len("format:"):] try: transformed_row[key] = datetime.strptime(row[key], format_str) except ValueError as e: logger.warning(f"Failed to format date {row[key]} with {format_str}: {e}") transformed_row[key] = row[key] else: transformed_row[key] = row[key] transformed.append(transformed_row) return transformed def load_to_sqlite(data: List[Dict[str, Any]], db_path: str, table_name: str) -> None: """ Load data into an SQLite database. Args: data: List of dictionaries to insert. db_path: Path to SQLite database file. table_name: Name of the table to insert data into. Raises: Exception: If database operations fail. """ try: conn = sqlite3.connect(db_path) cursor = conn.cursor() # Create table if not exists if data: columns = list(data[0].keys()) create_table_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join([f'{col} TEXT' for col in columns])})" cursor.execute(create_table_sql) # Insert data placeholders = ", ".join("?" * len(columns)) insert_sql = f"INSERT INTO {table_name} VALUES ({placeholders})" values = [tuple(row[col] for col in columns) for row in data] cursor.executemany(insert_sql, values) conn.commit() except Exception as e: logger.error(f"Failed to load data to SQLite: {e}") raise finally: if 'conn' in locals(): conn.close() def save_to_file(data: List[Dict[str, Any]], file_path: str, format: str = "json") -> None: """ Save data to a file. Args: data: List of dictionaries to save. file_path: Path to save the file. format: Format to save data ('json' or 'csv'). Raises: Exception: If file cannot be written. """ try: if format == "json": with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2) elif format == "csv": if data: columns = list(data[0].keys()) with open(file_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=columns) writer.writeheader() writer.writerows(data) else: raise ValueError(f"Unsupported format: {format}") except Exception as e: logger.error(f"Failed to save data to file: {e}") raise def run_pipeline( source_type: str, source_path: str, transformations: Dict[str, str], target_type: str, target_path: str, format: str = "json" ) -> None: """ Run the data pipeline. Args: source_type: Type of source ('csv', 'json', 'api'). source_path: Path or URL to the source. transformations: Mapping of field names to transformation rules. target_type: Type of target ('sqlite', 'file'). target_path: Path to the target location. format: Format to save if target is file ('json' or 'csv'). """ try: logger.info(f"Reading data from {source_type} source: {source_path}") if source_type == "csv": data = read_csv(source_path) elif source_type == "json": data = read_json(source_path) elif source_type == "api": data = read_api(source_path) else: raise ValueError(f"Unsupported source type: {source_type}") logger.info(f"Applying transformations: {transformations}") transformed_data = apply_transformations(data, transformations) logger.info(f"Loading data to {target_type} target: {target_path}") if target_type == "sqlite": load_to_sqlite(transformed_data, target_path, "data") elif target_type == "file": save_to_file(transformed_data, target_path, format) else: raise ValueError(f"Unsupported target type: {target_type}") logger.info("Pipeline completed successfully.") except Exception as e: logger.error(f"Pipeline failed: {e}") sys.exit(1) def main() -> None: """ Main function to parse command-line arguments and run the pipeline. """ parser = argparse.ArgumentParser(description="Data Pipeline ETL Tool") parser.add_argument("--source-type", required=True, choices=["csv", "json", "api"], help="Type of source (csv, json, api)") parser.add_argument("--source-path", required=True, help="Path or URL to the source") parser.add_argument("--transformations", required=False, help="Transformations as JSON string, e.g. '{\"name\": \"upper\"}'") parser.add_argument("--target-type", required=True, choices=["sqlite", "file"], help="Type of target (sqlite, file)") parser.add_argument("--target-path", required=True, help="Path to the target") parser.add_argument("--format", required=False, default="json", choices=["json", "csv"], help="Format for file output (default: json)") args = parser.parse_args() transformations: Dict[str, str] = {} if args.transformations: try: transformations = json.loads(args.transformations) except json.JSONDecodeError as e: logger.error(f"Invalid JSON for transformations: {e}") sys.exit(1) run_pipeline( source_type=args.source_type, source_path=args.source_path, transformations=transformations, target_type=args.target_type, target_path=args.target_path, format=args.format ) if __name__ == "__main__": main()