"""Application configuration.""" import logging import os from secrets_manager.python_ext import PythonSecretsManager from src.utils.private_key import get_private_key, get_private_key_from_file # Service information SERVICE_NAME = 'sound-recordings-takedown' # Production environment TEST_ENVIRONMENT = 'test' PROD_ENVIRONMENT = 'prod' DEV_ENVIRONMENT = 'dev' QA_ENVIRONMENT = 'qa' ENVIRONMENT = os.getenv('ENVIRONMENT', DEV_ENVIRONMENT) SENTRY_DSN = os.environ.get('SENTRY_DSN') secrets_manager_client = PythonSecretsManager( environment=ENVIRONMENT, service_name=SERVICE_NAME ) STORE_ID = os.getenv('STORE_ID') DELIVERY_TYPE = os.getenv('DELIVERY_TYPE') CONCURRENCY = int(os.getenv('CONCURRENCY', 100)) TIMEOUT = int(os.getenv('TIMEOUT', 300)) # Snowflake creds def load_private_key() -> bytes | None: """Load the Snowflake private key from secrets manager.""" private_key_passphrase = secrets_manager_client.get_cred( 'SNOWFLAKE_PRIVATE_KEY_PASSPHRASE' ) private_key_path = os.getenv('SNOWFLAKE_PRIVATE_KEY_PATH') if private_key_path: return get_private_key_from_file(private_key_path, private_key_passphrase) private_key = secrets_manager_client.get_cred('SNOWFLAKE_PRIVATE_KEY') if private_key: return get_private_key(private_key, private_key_passphrase) return None SNOWFLAKE_USER = os.getenv('SNOWFLAKE_USER') SNOWFLAKE_ACCOUNT = os.getenv('SNOWFLAKE_ACCOUNT') SNOWFLAKE_ROLE = os.getenv('SNOWFLAKE_ROLE') SNOWFLAKE_WAREHOUSE = os.getenv('SNOWFLAKE_WAREHOUSE') SNOWFLAKE_DATABASE = os.getenv('SNOWFLAKE_DATABASE') SNOWFLAKE_SCHEMA = os.getenv('SNOWFLAKE_SCHEMA') SNOWFLAKE_PRIVATE_KEY = load_private_key() # MySQL MYSQL_HOST = os.getenv('MYSQL_HOST') MYSQL_PORT = int(os.getenv('MYSQL_PORT', 3306)) MYSQL_USER = os.getenv('MYSQL_USER') MYSQL_PASSWORD = secrets_manager_client.get_cred( 'MYSQL_PASSWORD' ) MYSQL_DATABASE = os.getenv('MYSQL_DATABASE', 'art_relations') # S3 S3_BUCKET = os.getenv('S3_BUCKET', 'dev-orcdbucket') S3_TMP_FILE_DIR = os.getenv('S3_TMP_FILE_DIR', 'sound-recordings-takedown/') S3_EXPECTED_OWNER = os.getenv('S3_EXPECTED_OWNER', '103233932089') # AWS AWS_REGION = os.getenv('AWS_REGION', 'us-east-1') MAX_UPCS_COUNT = 2000 SEND_UPCS_TO_MASTERS_REGISTRY = os.getenv('SEND_UPCS_TO_MASTERS_REGISTRY', '').lower() == 'true' # Datadog datadog_secrets_manager_client = PythonSecretsManager( environment=ENVIRONMENT, service_name='datadog' ) DD_API_KEY = datadog_secrets_manager_client.get_cred('DD_API_KEY') DD_APP_KEY = datadog_secrets_manager_client.get_cred('DD_APP_KEY') # Logging LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO').upper() def setup_logger(name: str = SERVICE_NAME) -> logging.Logger: """Set up and return a configured logger.""" logger = logging.getLogger(name) if not logger.handlers: handler = logging.StreamHandler() handler.setFormatter(logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' )) logger.addHandler(handler) logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) return logger # E-mail configuration DEFAULT_EMAIL_SENDER = 'donotreply@theorchard.com' EMAIL_SENDER = os.getenv('EMAIL_SENDER', DEFAULT_EMAIL_SENDER) EMAIL_RECIPIENTS = os.getenv('EMAIL_RECIPIENTS', '').split(',')