nessi
117710cc0a
Replaced all inline error messages with the standardized `api_error` helper for consistent error response formatting. This improves clarity, maintainability, and ensures uniform error structures across the application. Updated logging for collector failures to include error class and switched to warning level for target unreachable scenarios.
64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
from functools import lru_cache
|
|
from pydantic import field_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
NEXAPG_VERSION = "0.2.0"
|
|
|
|
|
|
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
|
|
alert_rollback_ratio_window_minutes: int = 15
|
|
alert_rollback_ratio_min_total_transactions: int = 100
|
|
alert_rollback_ratio_min_rollbacks: int = 10
|
|
init_admin_email: str = "admin@example.com"
|
|
init_admin_password: str = "ChangeMe123!"
|
|
|
|
@property
|
|
def app_version(self) -> str:
|
|
return NEXAPG_VERSION
|
|
|
|
@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()
|