from __future__ import annotations import datetime import pathlib from typing import Any, Literal, Self from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization from fansifter_common.constants import PROD_ENVIRONMENT, QA_ENVIRONMENT from fansifter_common.utils.functional import lazy_proxy from pydantic import ( AliasChoices, Field, ImportString, ValidationError, field_validator, model_validator, ) from pydantic_core.core_schema import ValidationInfo from pydantic_settings import BaseSettings, SettingsConfigDict from snowflake.cortex import CompleteOptions from sqlalchemy.engine import URL from dmp.adapters.snowflake.converter import SnowflakeConverter from dmp.core.exceptions import ImproperlyConfiguredError class Settings(BaseSettings): # Main environment: str debug: bool = Field(validation_alias="APP_DEBUG", default=False) run_from_cli: bool = Field( validation_alias=AliasChoices("run_from_cli", "APP_RUN_FROM_CLI"), default=False ) # Directories base_dir: pathlib.Path = pathlib.Path(__file__).resolve().parent root_dir: pathlib.Path = base_dir.parent # Service service_name: str = "ows-dmp" service_version: str = "1.0.0" # AWS region aws_region_name: str = "us-east-1" # AWS Location service aws_location_api_key: str = Field("", validation_alias="AWS_LOCATION_API_KEY") # Postgres postgres_user: str = "postgres" postgres_password: str = "postgres" postgres_host: str = "localhost" postgres_port: int = 5432 postgres_name: str = "dmp" postgres_echo: bool = False postgres_pool_size: int = 10 postgres_pool_max_overflow: int = 20 postgres_pool_recycle: int = 4 * 60 * 60 # ~4 hours postgres_pool_pre_ping: bool = False postgres_pool_reset_on_return: Literal["commit", "rollback"] | None = None postgres_connect_args: dict[str, Any] = { "keepalives": 1, "keepalives_idle": 60, "keepalives_interval": 10, "keepalives_count": 5, } @property def postgres_url(self) -> URL: return URL.create( drivername="postgresql+psycopg", username=self.postgres_user, password=self.postgres_password, host=self.postgres_host, port=self.postgres_port, database=self.postgres_name, ) # Snowflake snowflake_account: str = "account" snowflake_database: str = "database" snowflake_schema: str = "" snowflake_warehouse: str = "warehouse" snowflake_role: str = "role" snowflake_user: str = "user" snowflake_password: str | None = None snowflake_host: str = "localhost" snowflake_port: int = 443 snowflake_url: str = "" @model_validator(mode="after") def _setup_snowflake_url(self) -> Any: from snowflake.sqlalchemy import URL self.snowflake_url = URL( account=self.snowflake_account, user=self.snowflake_user, database=self.snowflake_database, schema=self.snowflake_schema, warehouse=self.snowflake_warehouse, role=self.snowflake_role, ) return self snowflake_echo: bool = False snowflake_pool_size: int = 10 snowflake_pool_max_overflow: int = 20 snowflake_pool_recycle: int = 4 * 55 * 60 # ~4 hours snowflake_pool_pre_ping: bool = False snowflake_pool_reset_on_return: Literal["commit", "rollback"] | None = None default_snowflake_private_key_path: pathlib.Path = ( pathlib.Path.home() / ".ssh/snowflake/rsa_key.p8" ) snowflake_private_key_path: pathlib.Path = default_snowflake_private_key_path snowflake_private_key: bytes = b"" snowflake_key_passphrase: bytes = b"" snowflake_connect_args: dict[str, Any] = { "converter_class": SnowflakeConverter, "client_session_keep_alive": True, } @model_validator(mode="after") def _setup_snowflake_connect_args(self) -> Self: if self.environment == "test": return self # Load private key if not set and path exists if not self.snowflake_private_key and self.snowflake_private_key_path.exists(): self.snowflake_private_key = self.snowflake_private_key_path.read_bytes() # Check if private key and passphrase are set in QA/PROD environment if self.environment in (QA_ENVIRONMENT, PROD_ENVIRONMENT) and ( not self.snowflake_key_passphrase or not self.snowflake_private_key ): raise ValueError( "Snowflake private key passphrase and path must be set in " "QA/PROD environment." ) # Load private key if self.snowflake_private_key: pem_private_key = serialization.load_pem_private_key( self.snowflake_private_key, password=self.snowflake_key_passphrase or None, backend=default_backend(), ) private_key = pem_private_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) self.snowflake_connect_args["private_key"] = private_key return self snowflake_shopify_schema: str = "" snowflake_shopify_integrations_database: str = "QA_SHOPIFY_INTEGRATIONS" jinja2sql_template_searchpath: list[pathlib.Path] = [ base_dir / "auth" / "sql", base_dir / "ad_reporting" / "sql", base_dir / "fandata" / "sql", base_dir / "audiences" / "sql", base_dir / "artists" / "sql", base_dir / "shopify" / "sql", base_dir / "meta" / "sql", base_dir / "tiktok" / "sql", base_dir / "google" / "sql", base_dir / "rosters" / "sql", base_dir / "locations" / "sql", base_dir / "songwhip" / "sql", base_dir / "crm_campaigns" / "sql", base_dir / "crm_sublists" / "sql", ] # ORM models: list[ImportString[Any]] = [ "dmp.fandata.models", "dmp.locations.models", "dmp.rosters.models", "dmp.audiences.models", "dmp.shopify.models", "dmp.ad_accounts.models", "dmp.ad_reporting.models", "dmp.meta.models", "dmp.tiktok.models", "dmp.text_campaigns.models", ] # Redis redis_host: str = "localhost" redis_port: int = 6379 redis_ssl: bool = False @model_validator(mode="after") def _setup_redis_ssl(self) -> Self: if self.environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT]: self.redis_ssl = True return self # Cache cache_backend: str = "locmem" cache_default_timeout: int = 300 cache_key_prefix: str = "ows-dmp:" cache_redis_db: int = 0 @model_validator(mode="after") def _setup_cache_backend(self) -> Self: if self.environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT]: self.cache_backend = "redis" return self # M2M Token @property def m2m_token_secret_key_name(self) -> str: return f"{self.environment}/lambda-jwt-refresh/jwt_token" @property def m2m_token_secret_expiry_key_name(self) -> str: return f"{self.environment}/lambda-jwt-refresh/jwt_token_expiration" # Fivetran fivetran_api_key: str = "" fivetran_api_secret: str = "" fivetran_shopify_group_id: str = "" fivetran_reporting_group_id: str = "" fivetran_schema_prefix: str | None = None @model_validator(mode="after") def _setup_fivetran_schema_prefix(self) -> Self: if self.environment == "dev": self.fivetran_schema_prefix = self.snowflake_schema else: self.fivetran_schema_prefix = None return self # API jwt_auth_enabled: bool = True @model_validator(mode="after") def _setup_jwt_auth_enabled(self) -> Self: if self.environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT]: self.jwt_auth_enabled = True return self jwt_auth_exclude_paths: list[str] = [ "/hello/", "/redoc", "/docs*", "/openapi.json", "/_debug_toolbar", "/artists/participants", ] auth_allow_full_access: bool = False @model_validator(mode="after") def _setup_auth_allow_full_access(self) -> Self: """Allow to set full access authorization for local/dev environment(s) only.""" if self.environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT]: self.auth_allow_full_access = False if not self.auth_allow_full_access and not self.jwt_auth_enabled: raise ValueError( "For not enabled full access [AUTH_ALLOW_FULL_ACCESS], " "you must enable JWT authentication [JWT_AUTH_ENABLED=true]." ) return self # Testing @property def include_devtools_router(self) -> bool: return self.environment != PROD_ENVIRONMENT and not self.run_from_cli # OpenAPI / Swagger qa_server_url: str = "https://qa-ows-dmp.theorchard.io" openapi_url: str | None = "/openapi.json" @model_validator(mode="after") def _setup_openapi_url(self) -> Self: if self.environment == PROD_ENVIRONMENT: self.openapi_url = None return self openapi_filepath: pathlib.Path = root_dir / "spec/ows_dmp-1.0.0.yaml" openapi_servers: list[dict[str, Any]] | None = None @model_validator(mode="after") def _setup_openapi_servers(self) -> Self: if self.environment == "dev" and self.run_from_cli: self.openapi_servers = [{"url": self.qa_server_url}] return self # Split.io splitio_api_key: str = Field(validation_alias="SPLIT_IO_API_KEY", default="") splitio_block_until_ready_timeout: int = Field( validation_alias="SPLIT_IO_BLOCK_UNTIL_READY_TIMEOUT", default=60 ) splitio_config: dict[str, Any] = { "impressionsMode": "optimized", "preforkedInitialization": False, } @model_validator(mode="after") def _setup_splitio_config(self) -> Self: if self.splitio_api_key == "localhost": self.splitio_config.update( { "splitFile": str(self.root_dir / "tests/split.yaml"), } ) return self # Account access account_access_cache_timeout: int = 60 * 10 # 10 minutes # Artists artist_cache_prefix: str = "artist" artist_cache_reset_at: datetime.time = datetime.time( hour=6, minute=30, tzinfo=datetime.UTC ) # Audiences audience_global_excluded_countries: list[str] = [] audience_ads_excluded_countries: list[str] = [ "HU", "IN", "CN", ] audience_text_allowed_countries: list[str] = [ "US", "CA", "GB", "AU", "DE", ] @model_validator(mode="after") def _setup_audience_text_allowed_countries(self) -> Self: if self.environment != PROD_ENVIRONMENT: self.audience_text_allowed_countries.extend( [ "EE", "PL", "HR", "RO", "UA", ] ) return self audience_email_to_fansifter_only: bool = True @model_validator(mode="after") def _setup_audience_email_to_fansifter_only(self) -> Self: if self.environment == PROD_ENVIRONMENT: self.audience_email_to_fansifter_only = False return self # Audience Export audience_export_bucket_name: str = "" audience_export_stage: str = "audience_export_stage" # Audience cache audience_cache_timeout: int = 60 * 60 * 10 # 10 hours # Snowflake LLM configs segment_explain_llm_model: str = "llama3.1-70b" segment_explain_llm_options: CompleteOptions = { "temperature": 0.01, "guardrails": True, } segment_explain_llm_timeout: int = 15 # Logging logging_debug: bool = Field(False, validation_alias="LOGGING_DEBUG") @property def logging_config(self) -> dict[str, Any]: return { "version": 1, "disable_existing_loggers": True, "formatters": { "json": { "()": "owslogger.logger.DDJsonFormatter", "service_name": self.service_name, "service_version": self.service_version, "env": self.environment, }, "console": { "()": "fansifter_common.logging.ConsoleFormatter", }, }, "filters": { "require_debug_true": { "()": "fansifter_common.logging.RequireDebugTrueFilter", "value": self.logging_debug, }, "require_debug_false": { "()": "fansifter_common.logging.RequireDebugFalseFilter", "value": self.logging_debug, }, }, "handlers": { "stream": { "level": "INFO", "class": "logging.StreamHandler", "formatter": "json", "filters": ["require_debug_false"], }, "rich": { "level": "INFO", "class": "rich.logging.RichHandler", "filters": ["require_debug_true"], "formatter": "console", }, "null": { "level": "INFO", "class": "logging.NullHandler", }, }, "loggers": { "dmp": { "handlers": ["stream", "rich"], "level": "INFO", }, "fansifter_common.auth": { "handlers": ["stream", "rich"], "level": "WARNING", }, "fansifter_common.httpclient": { "handlers": ["stream", "rich"], "level": "WARNING", }, "fansifter_common.logging.asgi.middleware": { "handlers": ["stream", "rich"], "level": "INFO", }, "sqlalchemy.engine": { "handlers": ["stream", "rich"], "level": "WARNING", }, "httpx": { "handlers": ["stream", "rich"], "level": "WARNING", }, "uvicorn": { "handlers": ["stream", "rich"], "level": "INFO", }, "uvicorn.access": { "handlers": ["null"], "level": "INFO", }, }, } # Songwhip songwhip_url: str = "https://staging.songwhip.com" # Sentry sentry_dsn: str | None = None @model_validator(mode="after") def _setup_sentry_dsn(self) -> Self: if ( not self.sentry_dsn and self.environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT] and not self.run_from_cli ): raise ValueError(f"Sentry is not configured in {self.environment}.") return self # KMS kms_enabled: bool = True @field_validator("kms_enabled") @classmethod def _setup_kms_enabled(cls, v: bool, info: ValidationInfo) -> bool: environment: str = info.data["environment"] if environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT]: return True return v kms_key_id: str = Field(default="", validation_alias="DMP_KMS_KEY_ID") @field_validator("kms_key_id") @classmethod def _setup_kms_key_id(cls, v: str, info: ValidationInfo) -> str: kms_enabled: bool = info.data["kms_enabled"] if kms_enabled and not v: raise ValueError("`kms_key_id` must be set.") return v # Meta facebook_app_id: str = "" facebook_app_secret: str = "" meta_user_required_scopes: list[str] = [ "email", "public_profile", "ads_management", "business_management", ] meta_user_default_token_expires_in: int = 60 * 60 * 24 * 60 # 2 months campaign_assets_bucket_name: str = "qa-audience-campaign-assets" meta_business_accounts_cdn_assets_path: str = "meta-business-accounts" # TikTok tiktok_app_id: str = "" tiktok_app_secret: str = "" tiktok_user_required_scopes: list[int] = [ 18000000, 1, 3, 200, ] # Google google_app_id: str = "" google_app_secret: str = "" google_developer_token: str = "" google_user_required_scopes: list[str] = [ "https://www.googleapis.com/auth/adwords", "https://www.googleapis.com/auth/userinfo.profile", ] assets_cdn_domain: str = "" debug_toolbar_enabled: bool = Field( default=False, validation_alias="DEBUG_TOOLBAR_ENABLED" ) @field_validator("debug_toolbar_enabled") @classmethod def _setup_debug_toolbar_enabled(cls, v: bool, info: ValidationInfo) -> bool: environment: str = info.data["environment"] if environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT]: return False return v model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="allow", ) def setup_settings(**defaults: Any) -> Settings: """Initialize the application settings.""" try: return Settings(**defaults) except ValidationError as exc: errors = "\n".join( [ f" * {'.'.join(map(str, error['loc']))} - {error['msg']}" for error in exc.errors() ] ) raise ImproperlyConfiguredError( f"Failed to initialize settings:\n {errors}" ) from exc # Lazy proxy for settings settings = lazy_proxy(setup_settings)