chy88
ab1bd90c65
- 搭建 FastAPI、Redis、PostgreSQL、Nginx 与 Docker Compose 基础结构 - 实现反向代理、首用绑定、拦截告警、归档任务和管理接口 - 提供 Vue3 管理后台初版，以及 uv/requirements 双依赖配置
99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from functools import cached_property
|
|
from ipaddress import ip_network
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
RUNTIME_SETTINGS_REDIS_KEY = "sentinel:settings"
|
|
|
|
|
|
class RuntimeSettings(BaseModel):
|
|
alert_webhook_url: str | None = None
|
|
alert_threshold_count: int = Field(default=5, ge=1)
|
|
alert_threshold_seconds: int = Field(default=300, ge=1)
|
|
archive_days: int = Field(default=90, ge=1)
|
|
failsafe_mode: Literal["open", "closed"] = "closed"
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env",
|
|
env_file_encoding="utf-8",
|
|
extra="ignore",
|
|
case_sensitive=False,
|
|
)
|
|
|
|
downstream_url: str = Field(alias="DOWNSTREAM_URL")
|
|
redis_addr: str = Field(alias="REDIS_ADDR")
|
|
redis_password: str = Field(default="", alias="REDIS_PASSWORD")
|
|
pg_dsn: str = Field(alias="PG_DSN")
|
|
sentinel_hmac_secret: str = Field(alias="SENTINEL_HMAC_SECRET", min_length=32)
|
|
admin_password: str = Field(alias="ADMIN_PASSWORD", min_length=8)
|
|
admin_jwt_secret: str = Field(alias="ADMIN_JWT_SECRET", min_length=16)
|
|
trusted_proxy_ips: tuple[str, ...] = Field(default_factory=tuple, alias="TRUSTED_PROXY_IPS")
|
|
sentinel_failsafe_mode: Literal["open", "closed"] = Field(
|
|
default="closed",
|
|
alias="SENTINEL_FAILSAFE_MODE",
|
|
)
|
|
app_port: int = Field(default=7000, alias="APP_PORT")
|
|
alert_webhook_url: str | None = Field(default=None, alias="ALERT_WEBHOOK_URL")
|
|
alert_threshold_count: int = Field(default=5, alias="ALERT_THRESHOLD_COUNT", ge=1)
|
|
alert_threshold_seconds: int = Field(default=300, alias="ALERT_THRESHOLD_SECONDS", ge=1)
|
|
archive_days: int = Field(default=90, alias="ARCHIVE_DAYS", ge=1)
|
|
|
|
redis_binding_ttl_seconds: int = 604800
|
|
downstream_max_connections: int = 512
|
|
downstream_max_keepalive_connections: int = 128
|
|
last_used_flush_interval_seconds: int = 5
|
|
last_used_queue_size: int = 10000
|
|
login_lockout_threshold: int = 5
|
|
login_lockout_seconds: int = 900
|
|
admin_jwt_expire_hours: int = 8
|
|
archive_job_interval_minutes: int = 60
|
|
archive_batch_size: int = 500
|
|
metrics_ttl_days: int = 30
|
|
webhook_timeout_seconds: int = 5
|
|
|
|
@field_validator("downstream_url")
|
|
@classmethod
|
|
def normalize_downstream_url(cls, value: str) -> str:
|
|
return value.rstrip("/")
|
|
|
|
@field_validator("trusted_proxy_ips", mode="before")
|
|
@classmethod
|
|
def split_proxy_ips(cls, value: object) -> tuple[str, ...]:
|
|
if value is None:
|
|
return tuple()
|
|
if isinstance(value, str):
|
|
parts = [item.strip() for item in value.split(",")]
|
|
return tuple(item for item in parts if item)
|
|
if isinstance(value, (list, tuple, set)):
|
|
return tuple(str(item).strip() for item in value if str(item).strip())
|
|
return (str(value).strip(),)
|
|
|
|
@cached_property
|
|
def trusted_proxy_networks(self):
|
|
return tuple(ip_network(item, strict=False) for item in self.trusted_proxy_ips)
|
|
|
|
def build_runtime_settings(self) -> RuntimeSettings:
|
|
return RuntimeSettings(
|
|
alert_webhook_url=self.alert_webhook_url or None,
|
|
alert_threshold_count=self.alert_threshold_count,
|
|
alert_threshold_seconds=self.alert_threshold_seconds,
|
|
archive_days=self.archive_days,
|
|
failsafe_mode=self.sentinel_failsafe_mode,
|
|
)
|
|
|
|
|
|
_settings: Settings | None = None
|
|
|
|
|
|
def get_settings() -> Settings:
|
|
global _settings
|
|
if _settings is None:
|
|
_settings = Settings()
|
|
return _settings
|