173 lines
6.8 KiB
Python
173 lines
6.8 KiB
Python
"""Application configuration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from urllib.parse import quote_plus
|
|
|
|
from dotenv import load_dotenv
|
|
from pydantic import Field, computed_field, model_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
# Load .env file from project root
|
|
env_file = Path(__file__).parent.parent.parent / ".env"
|
|
if env_file.exists():
|
|
load_dotenv(env_file)
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
app_name: str = "PortfolioHub"
|
|
app_version: str = "0.1.1"
|
|
environment: str = "development"
|
|
allowed_origins_raw: str = Field(default="http://localhost:3000", validation_alias="PORTFOLIOHUB_ALLOWED_ORIGINS")
|
|
bootstrap_username: str = Field(default="", validation_alias="PORTFOLIOHUB_BOOTSTRAP_USERNAME")
|
|
bootstrap_email: str = Field(default="", validation_alias="PORTFOLIOHUB_BOOTSTRAP_EMAIL")
|
|
bootstrap_password: str = Field(default="", validation_alias="PORTFOLIOHUB_BOOTSTRAP_PASSWORD")
|
|
read_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_READ_TOKENS")
|
|
trade_write_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_TRADE_WRITE_TOKENS")
|
|
admin_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_ADMIN_TOKENS")
|
|
db_enabled: bool = Field(default=False, validation_alias="PORTFOLIOHUB_DB_ENABLED")
|
|
session_token_ttl_hours: int = Field(default=12, validation_alias="PORTFOLIOHUB_SESSION_TOKEN_TTL_HOURS")
|
|
client_token_default_ttl_days: int = Field(default=365, validation_alias="PORTFOLIOHUB_CLIENT_TOKEN_DEFAULT_TTL_DAYS")
|
|
bot_token_default_ttl_days: int = Field(default=365, validation_alias="PORTFOLIOHUB_BOT_TOKEN_DEFAULT_TTL_DAYS")
|
|
|
|
# External services configuration
|
|
timescaledb_host: str = Field(default="localhost", validation_alias="TIMESCALEDB_HOST")
|
|
timescaledb_port: int = Field(default=5432, validation_alias="TIMESCALEDB_PORT")
|
|
timescaledb_name: str = Field(default="portfoliohub", validation_alias="TIMESCALEDB_NAME")
|
|
timescaledb_user: str = Field(default="portfoliohub", validation_alias="TIMESCALEDB_USER")
|
|
timescaledb_password: str = Field(default="", validation_alias="TIMESCALEDB_PASSWORD")
|
|
timescaledb_sslmode: str = Field(default="prefer", validation_alias="TIMESCALEDB_SSLMODE")
|
|
timescaledb_connect_timeout_sec: int = Field(default=10, validation_alias="TIMESCALEDB_CONNECT_TIMEOUT_SEC")
|
|
timescaledb_application_name: str = Field(default="portfoliohub", validation_alias="TIMESCALEDB_APPLICATION_NAME")
|
|
|
|
redis_host: str = Field(default="localhost", validation_alias="REDIS_HOST")
|
|
redis_port: int = Field(default=6379, validation_alias="REDIS_PORT")
|
|
redis_db: int = Field(default=0, validation_alias="REDIS_DB")
|
|
redis_password: str = Field(default="", validation_alias="REDIS_PASSWORD")
|
|
|
|
loki_host: str = Field(default="localhost", validation_alias="LOKI_HOST")
|
|
loki_port: int = Field(default=3100, validation_alias="LOKI_PORT")
|
|
loki_enabled: bool = Field(default=False, validation_alias="LOKI_ENABLED")
|
|
credential_encryption_key: str = Field(default="", validation_alias="CREDENTIAL_ENCRYPTION_KEY")
|
|
root_path: str = Field(default="", validation_alias="PORTFOLIOHUB_ROOT_PATH")
|
|
|
|
model_config = SettingsConfigDict(
|
|
env_prefix="PORTFOLIOHUB_",
|
|
case_sensitive=False,
|
|
extra="ignore",
|
|
populate_by_name=True,
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_credential_encryption_key(self) -> "Settings":
|
|
# Only validate at runtime when the key is actually used.
|
|
# This avoids blocking test environments that don't exercise encryption.
|
|
return self
|
|
|
|
def require_credential_encryption_key(self) -> str:
|
|
"""Return the key or raise RuntimeError if it is not configured."""
|
|
if not self.credential_encryption_key.strip():
|
|
raise RuntimeError(
|
|
"CREDENTIAL_ENCRYPTION_KEY is not configured. "
|
|
"Generate one with: python -c \"from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())\""
|
|
)
|
|
return self.credential_encryption_key
|
|
|
|
@computed_field
|
|
@property
|
|
def allowed_origins(self) -> list[str]:
|
|
return [item.strip() for item in self.allowed_origins_raw.split(",") if item.strip()]
|
|
|
|
@computed_field
|
|
@property
|
|
def bootstrap_user_username(self) -> str | None:
|
|
normalized = self.bootstrap_username.strip()
|
|
return normalized or None
|
|
|
|
@computed_field
|
|
@property
|
|
def bootstrap_user_email(self) -> str | None:
|
|
normalized = self.bootstrap_email.strip()
|
|
return normalized or None
|
|
|
|
@computed_field
|
|
@property
|
|
def bootstrap_user_password(self) -> str | None:
|
|
normalized = self.bootstrap_password.strip()
|
|
return normalized or None
|
|
|
|
@computed_field
|
|
@property
|
|
def read_tokens(self) -> set[str]:
|
|
return {item.strip() for item in self.read_tokens_raw.split(",") if item.strip()}
|
|
|
|
@computed_field
|
|
@property
|
|
def trade_write_tokens(self) -> set[str]:
|
|
return {item.strip() for item in self.trade_write_tokens_raw.split(",") if item.strip()}
|
|
|
|
@computed_field
|
|
@property
|
|
def admin_tokens(self) -> set[str]:
|
|
return {item.strip() for item in self.admin_tokens_raw.split(",") if item.strip()}
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_host(self) -> str:
|
|
return self.timescaledb_host
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_port(self) -> int:
|
|
return self.timescaledb_port
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_name(self) -> str:
|
|
return self.timescaledb_name
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_user(self) -> str:
|
|
return self.timescaledb_user
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_password(self) -> str:
|
|
return self.timescaledb_password
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_sslmode(self) -> str:
|
|
return self.timescaledb_sslmode
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_connect_timeout_sec(self) -> int:
|
|
return self.timescaledb_connect_timeout_sec
|
|
|
|
@computed_field
|
|
@property
|
|
def active_db_application_name(self) -> str:
|
|
return self.timescaledb_application_name
|
|
|
|
@computed_field
|
|
@property
|
|
def active_database_url(self) -> str:
|
|
user = quote_plus(self.active_db_user)
|
|
password = quote_plus(self.active_db_password)
|
|
database = quote_plus(self.active_db_name)
|
|
return (
|
|
f"postgresql+psycopg://{user}:{password}@{self.active_db_host}:{self.active_db_port}/{database}"
|
|
f"?sslmode={quote_plus(self.active_db_sslmode)}"
|
|
f"&connect_timeout={self.active_db_connect_timeout_sec}"
|
|
f"&application_name={quote_plus(self.active_db_application_name)}"
|
|
)
|
|
|
|
|
|
@lru_cache
|
|
def get_settings() -> Settings:
|
|
return Settings()
|