""" Configuration management for the owned clip analysis backfill job. This module centralizes all configuration settings that were previously scattered throughout the monolithic script. """ from dataclasses import dataclass from datetime import datetime from typing import Optional import boto3 @dataclass class RetryConfig: """Configuration for retry logic and error handling.""" max_retries: int = 5 base_delay: float = 0.5 max_delay: float = 30.0 backoff_multiplier: float = 2.0 jitter_range: float = 0.1 @dataclass class BatchConfig: """Configuration for batch processing.""" batch_size: int = 500 max_batches_per_connection: int = 50 page_size: int = 1000 throttle_delay_seconds: float = 0.25 @dataclass class DatabaseConfig: """Configuration for database connections.""" transaction_isolation_level: str = "READ COMMITTED" connection_timeout: int = 30 query_timeout: int = 300 max_concurrent_connections: int = 10 @dataclass class S3Config: """Configuration for S3 operations.""" partition_prefix: str = "owned_clip_analysis_export/" partition_date_format: str = "%Y-%m-%d" @dataclass class CloudWatchConfig: """Configuration for CloudWatch metrics.""" namespace: str = "SunoETL/GlueJobs" region: str = "us-east-1" metrics_batch_size: int = 20 @dataclass class SparkConfig: """Configuration for Spark optimization.""" min_partitions: int = 10 max_partitions: int = 100 rows_per_partition_target: int = 500000 class JobConfig: """ Centralized configuration management for the backfill job. This class handles environment detection, validation, and provides type-safe access to all configuration settings. """ # Environment constants PROD_ACCOUNT_ID = "734185074900" STAGING_ACCOUNT_ID = "590183763515" def __init__(self, job_name: str, backfill_start_date: str, backfill_end_date: str, batch_size: Optional[int] = None, throttle_delay_seconds: Optional[float] = None, max_batches_per_connection: Optional[int] = None): """ Initialize job configuration with validation. Args: job_name: Name of the Glue job backfill_start_date: Start date in YYYY-MM-DD format backfill_end_date: End date in YYYY-MM-DD format batch_size: Optional batch size override throttle_delay_seconds: Optional throttle delay override max_batches_per_connection: Optional max batches per connection override Raises: ValueError: If dates are invalid or date range is invalid """ self.job_name = job_name self.backfill_start_date = self._validate_date(backfill_start_date) self.backfill_end_date = self._validate_date(backfill_end_date) # Validate date range if self.backfill_start_date >= self.backfill_end_date: raise ValueError( f"backfill_start_date ({backfill_start_date}) must be before " f"backfill_end_date ({backfill_end_date})" ) # Detect environment self._detect_environment() # Application name for database connections self.application_name = "snowflake_to_rds_owned_clip_analysis_backfill" # Initialize configuration components self.retry = RetryConfig() self.batch = BatchConfig( batch_size=batch_size if batch_size is not None else BatchConfig.batch_size, throttle_delay_seconds=throttle_delay_seconds if throttle_delay_seconds is not None else BatchConfig.throttle_delay_seconds, max_batches_per_connection=max_batches_per_connection if max_batches_per_connection is not None else BatchConfig.max_batches_per_connection, page_size=BatchConfig.page_size ) self.database = DatabaseConfig() self.s3 = S3Config() self.cloudwatch = CloudWatchConfig() self.spark = SparkConfig() def _validate_date(self, date_str: str) -> str: """ Validate date format. Args: date_str: Date string in YYYY-MM-DD format Returns: Validated date string Raises: ValueError: If date format is invalid """ try: datetime.strptime(date_str, "%Y-%m-%d") return date_str except ValueError as e: raise ValueError( f"Invalid date format. Expected format: YYYY-MM-DD. Error: {e}" ) def _detect_environment(self): """Detect the current AWS environment and set environment-specific settings.""" try: sts = boto3.client("sts") account_id = sts.get_caller_identity()["Account"] if account_id == self.PROD_ACCOUNT_ID: self.environment = "PROD" self.database_name = "SUNO_PROD" self.s3_bucket = "analytics-database-data" elif account_id == self.STAGING_ACCOUNT_ID: self.environment = "STAGING" self.database_name = "SUNO_STAGING" self.s3_bucket = "analytics-database-data-staging" else: raise ValueError(f"Invalid account ID: {account_id}") self.account_id = account_id except Exception as e: raise RuntimeError(f"Failed to detect AWS environment: {e}") @property def table_name(self) -> str: """Get the source table name.""" return "OWNED_CLIP_ANALYSIS" @property def target_table_name(self) -> str: """Get the target RDS table name.""" return "billing_clipownership" @property def s3_partition_prefix(self) -> str: """Get the full S3 partition prefix path.""" return f"s3://{self.s3_bucket}/{self.s3.partition_prefix}" def get_target_partitions(self, total_rows: int) -> int: """ Calculate optimal number of Spark partitions based on data size. Args: total_rows: Total number of rows to process Returns: Optimal number of partitions """ target_partitions = 1 # If we want to use partitioning, uncomment the following code # target_partitions = max( # self.spark.min_partitions, # min(self.spark.max_partitions, total_rows // self.spark.rows_per_partition_target) # ) return target_partitions def validate(self): """ Perform comprehensive configuration validation. Raises: ValueError: If any configuration is invalid """ # Validate retry configuration if self.retry.max_retries < 1: raise ValueError("max_retries must be at least 1") if self.retry.base_delay <= 0: raise ValueError("base_delay must be positive") # Validate batch configuration if self.batch.batch_size < 1: raise ValueError("batch_size must be at least 1") if self.batch.max_batches_per_connection < 1: raise ValueError("max_batches_per_connection must be at least 1") # Validate Spark configuration if self.spark.min_partitions < 1: raise ValueError("min_partitions must be at least 1") if self.spark.max_partitions < self.spark.min_partitions: raise ValueError("max_partitions must be >= min_partitions") def to_dict(self) -> dict: """ Convert configuration to a serializable dictionary for Spark distribution. Returns: Dictionary containing all serializable configuration data """ return { "job_name": self.job_name, "backfill_start_date": self.backfill_start_date, "backfill_end_date": self.backfill_end_date, "environment": self.environment, "account_id": self.account_id, "database_name": self.database_name, "s3_bucket": self.s3_bucket, # Nested configs as dicts "retry": { "max_retries": self.retry.max_retries, "base_delay": self.retry.base_delay, "max_delay": self.retry.max_delay, "backoff_multiplier": self.retry.backoff_multiplier, "jitter_range": self.retry.jitter_range }, "batch": { "batch_size": self.batch.batch_size, "max_batches_per_connection": self.batch.max_batches_per_connection, "page_size": self.batch.page_size, "throttle_delay_seconds": self.batch.throttle_delay_seconds }, "database": { "transaction_isolation_level": self.database.transaction_isolation_level, "connection_timeout": self.database.connection_timeout, "query_timeout": self.database.query_timeout, "max_concurrent_connections": self.database.max_concurrent_connections }, "s3": { "partition_prefix": self.s3.partition_prefix, "partition_date_format": self.s3.partition_date_format }, "cloudwatch": { "namespace": self.cloudwatch.namespace, "region": self.cloudwatch.region, "metrics_batch_size": self.cloudwatch.metrics_batch_size }, "spark": { "min_partitions": self.spark.min_partitions, "max_partitions": self.spark.max_partitions, "rows_per_partition_target": self.spark.rows_per_partition_target }, "application_name": self.application_name } @classmethod def from_dict(cls, config_dict: dict): """ Reconstruct JobConfig from a serializable dictionary. Args: config_dict: Dictionary created by to_dict() Returns: JobConfig instance """ # Create base instance instance = cls.__new__(cls) # Set basic properties instance.job_name = config_dict["job_name"] instance.backfill_start_date = config_dict["backfill_start_date"] instance.backfill_end_date = config_dict["backfill_end_date"] instance.environment = config_dict["environment"] instance.account_id = config_dict["account_id"] instance.database_name = config_dict["database_name"] instance.s3_bucket = config_dict["s3_bucket"] instance.application_name = config_dict["application_name"] # Reconstruct nested config objects instance.retry = RetryConfig(**config_dict["retry"]) instance.batch = BatchConfig(**config_dict["batch"]) instance.database = DatabaseConfig(**config_dict["database"]) instance.s3 = S3Config(**config_dict["s3"]) instance.cloudwatch = CloudWatchConfig(**config_dict["cloudwatch"]) instance.spark = SparkConfig(**config_dict["spark"]) return instance def __repr__(self) -> str: """Return a string representation of the configuration.""" return ( f"JobConfig(" f"environment={self.environment}, " f"job_name={self.job_name}, " f"date_range={self.backfill_start_date} to {self.backfill_end_date}, " f"s3_bucket={self.s3_bucket}" f")" )