import secrets import warnings from typing import Annotated, Any, Literal from urllib.parse import parse_qs, unquote from pydantic import ( AnyUrl, BeforeValidator, GetCoreSchemaHandler, GetJsonSchemaHandler, HttpUrl, PostgresDsn, computed_field, ) from pydantic import v1 as pydantic_v1 from pydantic_core import CoreSchema, core_schema from pydantic_settings import BaseSettings, SettingsConfigDict from typing_extensions import Self def parse_cors(v: Any) -> list[str] | str: if isinstance(v, str) and not v.startswith("["): return [i.strip() for i in v.split(",")] elif isinstance(v, list | str): return v raise ValueError(v) class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_ignore_empty=True, extra="ignore" ) API_V1_STR: str = "/api/v1" SECRET_KEY: str = secrets.token_urlsafe(32) # 60 minutes * 24 hours * 8 days = 8 days ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 DOMAIN: str = "localhost" ENVIRONMENT: Literal["local", "staging", "production"] = "local" @computed_field # type: ignore[misc] @property def server_host(self) -> str: # Use HTTPS for anything other than local development if self.ENVIRONMENT == "local": return f"http://{self.DOMAIN}" return f"https://{self.DOMAIN}" BACKEND_CORS_ORIGINS: Annotated[ list[AnyUrl] | str, BeforeValidator(parse_cors) ] = [] PROJECT_NAME: str SENTRY_DSN: HttpUrl | None = None POSTGRES_SERVER: str POSTGRES_PORT: int = 5432 POSTGRES_USER: str POSTGRES_PASSWORD: str POSTGRES_DB: str = "" @computed_field # type: ignore[misc] @property def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn: return Url.build( scheme="postgresql+psycopg", user=self.POSTGRES_USER, password=self.POSTGRES_PASSWORD, host=self.POSTGRES_SERVER, port=self.POSTGRES_PORT, path=self.POSTGRES_DB, ) PROJECT_NAME: str SENTRY_DSN: HttpUrl | None = None def _check_default_secret(self, var_name: str, value: str | None) -> None: if value == "changethis": message = ( f'The value of {var_name} is "changethis", ' "for security, please change it, at least for deployments." ) if self.ENVIRONMENT == "local": warnings.warn(message, stacklevel=1) else: raise ValueError(message) class Url(pydantic_v1.AnyUrl): quoted: bool = False def query_params(self) -> dict[str, str]: return parse_qs(self.query) @property def username(self) -> str | None: return self.user @classmethod def build( cls, *, scheme: str, user: str | None = None, password: str | None = None, host: str | None = None, port: int | None = None, path: str | None = None, query: str | None = None, fragment: str | None = None, **_kwargs: str, ) -> Self: return super().build( scheme=scheme, user=user, password=password, host=host if host is not None else "", port=str(port) if port is not None else None, path=path, query=query, fragment=fragment, **_kwargs, ) def _stringify_url(self) -> str: return self.build( scheme=self.scheme, user=unquote(self.username) if self.username is not None else None, password=unquote(self.password) if self.password is not None else None, host=self.host, port=self.port, path=self.path, query=unquote(self.query) if self.query is not None else None, fragment=self.fragment, ) @classmethod def _validate_from_str(cls, value: str): if cls.strip_whitespace: value = value.strip() m = cls._match_url(value) # the regex should always match, # if it doesn't please report with details of the URL tried if m is None: raise ValueError("URL regex failed unexpectedly") original_parts = m.groupdict() parts = cls.apply_default_parts(original_parts) parts = cls.validate_parts(parts) if m.end() != len(value): raise ValueError( "URL invalid, extra characters found after valid URL:" f" {value[m.end() :]}" ) return cls._build_url(m, value, parts) @classmethod def __get_pydantic_core_schema__( cls, _source_type: Any, _handler: GetCoreSchemaHandler, ) -> core_schema.CoreSchema: from_str_schema = core_schema.chain_schema( [ core_schema.str_schema( max_length=cls.max_length, min_length=cls.min_length, ), core_schema.no_info_plain_validator_function(cls._validate_from_str), ] ) return core_schema.json_or_python_schema( json_schema=from_str_schema, python_schema=from_str_schema, serialization=core_schema.plain_serializer_function_ser_schema( lambda instance: pydantic_v1.AnyUrl.__str__(instance) ), ) @classmethod def __get_pydantic_json_schema__( cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler, ) -> dict[str, Any]: json_schema = handler(core_schema) json_schema = handler.resolve_ref_schema(json_schema) super().__modify_schema__(json_schema) return json_schema def __str__(self): return ( self._stringify_url() if self.quoted else pydantic_v1.AnyUrl.__str__(self) ) settings = Settings() # type: ignore