from functools import lru_cache from os.path import abspath from os.path import dirname as pdirname from os.path import join as pjoin from pathlib import Path from generalresearch.models.custom_types import InfluxDsn from jb.models.custom_types import UUIDStr from pydantic import Field, HttpUrl, PostgresDsn, RedisDsn, SecretStr, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict BASE_DIR = pdirname(pdirname(abspath(__file__))) BASE_HTML_PATH = Path(BASE_DIR) / "templates" / "base.html" BASE_HTML = BASE_HTML_PATH.read_text() class AmtJbBaseSettings(BaseSettings): debug: bool = Field(default=True) redis: RedisDsn | None = Field(default=None) redis_timeout: float = Field(default=0.10) amt_jb_db: PostgresDsn | None = Field(default=None) amt_endpoint: HttpUrl | None = Field(default=None) amt_access_id: str | None = Field(default=None) amt_secret_key: str | None = Field(default=None) aws_owner_id: str | None = Field(default=None) aws_subscription_arn: str | None = Field(default=None) class Settings(AmtJbBaseSettings): model_config = SettingsConfigDict( env_file=( pjoin(BASE_DIR, x) for x in [".env.test", ".env.testing", ".env.staging", ".env.prod"] ), env_file_encoding="utf-8", case_sensitive=False, extra="allow", cli_parse_args=False, ) debug: bool = False app_name: str = "AMT JB API" base_url: HttpUrl = Field(default=HttpUrl("https://jamesbillings67.com/")) fsb_host: HttpUrl = Field(default=HttpUrl("https://fsb.generalresearch.com/")) # Needed for admin function on fsb w/o authentication fsb_host_private_route: str | None = Field(default=None) product_id: UUIDStr | None = Field(default=None) influx_db: InfluxDsn | None = Field(default=None) sns_path: str | None = Field(default=None) session_token_ttl_seconds: int = Field(default=30 * 24 * 60 * 60, gt=0) session_jwt_secret: SecretStr | None = Field(default=None, min_length=32) magic_token_salt: SecretStr | None = Field(default=None, min_length=32) gr_api_host: HttpUrl = Field(default=HttpUrl("https://generalresearch.com/api/v2/")) gr_api_token: SecretStr | None = Field(default=None, min_length=1) mautic_api_key: SecretStr | None = Field(default=None, min_length=32) @model_validator(mode="after") def validate_host_and_key(self) -> "Settings": if not self.amt_jb_db: raise ValueError("amt_jb_db is required") if not self.aws_owner_id: raise ValueError("aws_owner_id is required") if not self.aws_subscription_arn: raise ValueError("aws_subscription_arn is required") if not self.product_id: raise ValueError("product_id is required") if not self.sns_path: raise ValueError("sns_path is required") if not self.session_jwt_secret: raise ValueError("session_jwt_secret is required") if not self.magic_token_salt: raise ValueError("magic_token_salt is required") if self.session_jwt_secret == self.magic_token_salt: raise ValueError("JWT Secret must be different than Magic Token Salt") if not self.gr_api_token: raise ValueError("gr_api_token is required") if not self.mautic_api_key: raise ValueError("mautic_api_key is required") return self @lru_cache def get_settings(): return Settings()