mi
96 lines
4.3 KiB
Python
96 lines
4.3 KiB
Python
from functools import lru_cache
|
|
from typing import Literal
|
|
|
|
from pydantic import AnyHttpUrl, Field, SecretStr, model_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env", env_file_encoding="utf-8", case_sensitive=True, extra="ignore"
|
|
)
|
|
|
|
app_env: str = Field(alias="APP_ENV")
|
|
api_port: int = Field(default=8000, alias="API_PORT")
|
|
log_level: str = Field(default="INFO", alias="LOG_LEVEL")
|
|
database_url: str = Field(alias="DATABASE_URL")
|
|
redis_url: str = Field(alias="REDIS_URL")
|
|
redis_realtime_url: str = Field(alias="REDIS_REALTIME_URL")
|
|
|
|
keycloak_public_url: AnyHttpUrl = Field(alias="KEYCLOAK_PUBLIC_URL")
|
|
keycloak_internal_url: AnyHttpUrl = Field(alias="KEYCLOAK_INTERNAL_URL")
|
|
keycloak_realm: str = Field(alias="KEYCLOAK_REALM")
|
|
keycloak_audience: str = Field(alias="KEYCLOAK_AUDIENCE")
|
|
jwks_cache_ttl_seconds: int = 300
|
|
jwks_stale_grace_seconds: int = 900
|
|
|
|
message_safety_url: AnyHttpUrl = Field(alias="MESSAGE_SAFETY_URL")
|
|
message_safety_service_token: SecretStr = Field(alias="MESSAGE_SAFETY_SERVICE_TOKEN")
|
|
message_safety_ca_file: str | None = Field(default=None, alias="MESSAGE_SAFETY_CA_FILE")
|
|
message_safety_api_prefix: Literal["/internal/safety/v2"] = Field(
|
|
default="/internal/safety/v2", alias="MESSAGE_SAFETY_API_PREFIX"
|
|
)
|
|
message_safety_post_timeout_sec: float = Field(
|
|
default=5, alias="MESSAGE_SAFETY_POST_TIMEOUT_SEC"
|
|
)
|
|
message_safety_task_poll_interval_sec: float = Field(
|
|
default=2, alias="MESSAGE_SAFETY_TASK_POLL_INTERVAL_SEC"
|
|
)
|
|
message_safety_task_poll_max_sec: float = Field(
|
|
default=300, alias="MESSAGE_SAFETY_TASK_POLL_MAX_SEC"
|
|
)
|
|
message_safety_circuit_failure_threshold: int = Field(
|
|
default=5, alias="MESSAGE_SAFETY_CIRCUIT_FAILURE_THRESHOLD"
|
|
)
|
|
message_safety_circuit_open_sec: int = Field(
|
|
default=30, alias="MESSAGE_SAFETY_CIRCUIT_OPEN_SEC"
|
|
)
|
|
|
|
bitrix_local_app_base_url: AnyHttpUrl = Field(alias="BITRIX_LOCAL_APP_BASE_URL")
|
|
bitrix_local_app_internal_token: SecretStr = Field(alias="BITRIX_LOCAL_APP_INTERNAL_TOKEN")
|
|
bitrix_api_inbox_token: SecretStr = Field(alias="BITRIX_API_INBOX_TOKEN")
|
|
bitrix_local_app_http_timeout_sec: float = Field(
|
|
default=20, alias="BITRIX_LOCAL_APP_HTTP_TIMEOUT_SEC"
|
|
)
|
|
bitrix_local_app_circuit_failure_threshold: int = Field(
|
|
default=5, alias="BITRIX_LOCAL_APP_CIRCUIT_FAILURE_THRESHOLD"
|
|
)
|
|
bitrix_local_app_circuit_open_sec: int = Field(
|
|
default=30, alias="BITRIX_LOCAL_APP_CIRCUIT_OPEN_SEC"
|
|
)
|
|
keycloak_settings_bridge_token: SecretStr = Field(alias="KEYCLOAK_SETTINGS_BRIDGE_TOKEN")
|
|
|
|
selectel_s3_endpoint_url: AnyHttpUrl = Field(alias="SELECTEL_S3_ENDPOINT_URL")
|
|
selectel_s3_bucket_documents: str = Field(alias="SELECTEL_S3_BUCKET_DOCUMENTS")
|
|
selectel_s3_bucket_attachments: str = Field(alias="SELECTEL_S3_BUCKET_ATTACHMENTS")
|
|
selectel_s3_bucket_quarantine: str = Field(alias="SELECTEL_S3_BUCKET_QUARANTINE")
|
|
selectel_s3_access_key: SecretStr = Field(alias="SELECTEL_S3_ACCESS_KEY")
|
|
selectel_s3_secret_key: SecretStr = Field(alias="SELECTEL_S3_SECRET_KEY")
|
|
otel_exporter_otlp_endpoint: str | None = Field(
|
|
default=None, alias="OTEL_EXPORTER_OTLP_ENDPOINT"
|
|
)
|
|
cursor_hmac_secret: SecretStr = Field(alias="CURSOR_HMAC_SECRET")
|
|
trusted_proxy_cidrs: str = Field(default="127.0.0.1/32", alias="TRUSTED_PROXY_CIDRS")
|
|
worker_poll_interval_sec: float = Field(default=2, alias="WORKER_POLL_INTERVAL_SEC")
|
|
notifications_token_producer_test: SecretStr | None = Field(
|
|
default=None, alias="NOTIFICATIONS_TOKEN_PRODUCER_TEST"
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def require_safety_tls_in_deployed_environments(self) -> "Settings":
|
|
if self.app_env not in {"local", "test"}:
|
|
if str(self.message_safety_url).split(":", 1)[0] != "https":
|
|
raise ValueError("MESSAGE_SAFETY_URL must use HTTPS")
|
|
if not self.message_safety_ca_file:
|
|
raise ValueError("MESSAGE_SAFETY_CA_FILE is required")
|
|
return self
|
|
|
|
@property
|
|
def issuer(self) -> str:
|
|
return f"{str(self.keycloak_public_url).rstrip('/')}/realms/{self.keycloak_realm}"
|
|
|
|
|
|
@lru_cache
|
|
def get_settings() -> Settings:
|
|
return Settings()
|