All checks were successful
PostgreSQL Compatibility Matrix / PG14 smoke (push) Successful in 8s
PostgreSQL Compatibility Matrix / PG15 smoke (push) Successful in 7s
PostgreSQL Compatibility Matrix / PG16 smoke (push) Successful in 7s
PostgreSQL Compatibility Matrix / PG17 smoke (push) Successful in 7s
PostgreSQL Compatibility Matrix / PG18 smoke (push) Successful in 7s
Introduced `ALERT_ACTIVE_CONNECTION_RATIO_MIN_TOTAL_CONNECTIONS` to reduce false positives by requiring a minimum number of total connections before evaluating the active connection ratio. Updated the logic and description in relevant files for clarity and configurability.
55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
from functools import lru_cache
|
|
from pydantic import field_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
|
|
|
|
app_name: str = "NexaPG Monitor"
|
|
environment: str = "dev"
|
|
api_v1_prefix: str = "/api/v1"
|
|
log_level: str = "INFO"
|
|
|
|
db_host: str = "db"
|
|
db_port: int = 5432
|
|
db_name: str = "nexapg"
|
|
db_user: str = "nexapg"
|
|
db_password: str = "nexapg"
|
|
|
|
jwt_secret_key: str
|
|
jwt_algorithm: str = "HS256"
|
|
jwt_access_token_minutes: int = 15
|
|
jwt_refresh_token_minutes: int = 60 * 24 * 7
|
|
|
|
encryption_key: str
|
|
cors_origins: str = "http://localhost:5173"
|
|
poll_interval_seconds: int = 30
|
|
alert_active_connection_ratio_min_total_connections: int = 5
|
|
init_admin_email: str = "admin@example.com"
|
|
init_admin_password: str = "ChangeMe123!"
|
|
|
|
@property
|
|
def database_url(self) -> str:
|
|
return (
|
|
f"postgresql+asyncpg://{self.db_user}:{self.db_password}"
|
|
f"@{self.db_host}:{self.db_port}/{self.db_name}"
|
|
)
|
|
|
|
@property
|
|
def cors_origins_list(self) -> list[str]:
|
|
return [item.strip() for item in self.cors_origins.split(",") if item.strip()]
|
|
|
|
@field_validator("environment")
|
|
@classmethod
|
|
def validate_environment(cls, value: str) -> str:
|
|
allowed = {"dev", "staging", "prod", "test"}
|
|
if value not in allowed:
|
|
raise ValueError(f"environment must be one of {allowed}")
|
|
return value
|
|
|
|
|
|
@lru_cache
|
|
def get_settings() -> Settings:
|
|
return Settings()
|