373 lines
15 KiB
Python
373 lines
15 KiB
Python
"""
|
|
Application Settings and Configuration Management
|
|
|
|
This module provides centralized configuration management using Pydantic Settings.
|
|
All environment variables and application settings are defined here with proper
|
|
validation and type hints.
|
|
"""
|
|
|
|
from typing import Optional, Dict, Any, List
|
|
from functools import lru_cache
|
|
from pydantic import Field, field_validator, ConfigDict
|
|
from pydantic_settings import BaseSettings
|
|
from pathlib import Path
|
|
|
|
|
|
class DatabaseSettings(BaseSettings):
|
|
"""Database connection settings"""
|
|
|
|
host: str = Field(default="127.0.0.1", env="MYSQL_HOST")
|
|
port: int = Field(default=3306, env="MYSQL_PORT")
|
|
database: str = Field(default="stupa", env="MYSQL_DB")
|
|
username: str = Field(default="stupa", env="MYSQL_USER")
|
|
password: str = Field(default="secret", env="MYSQL_PASSWORD")
|
|
pool_size: int = Field(default=10, env="DB_POOL_SIZE")
|
|
max_overflow: int = Field(default=20, env="DB_MAX_OVERFLOW")
|
|
pool_pre_ping: bool = Field(default=True, env="DB_POOL_PRE_PING")
|
|
echo: bool = Field(default=False, env="DB_ECHO")
|
|
|
|
def __init__(self, **kwargs):
|
|
"""Initialize DatabaseSettings and log environment variables"""
|
|
import os
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
logger.info(f"DatabaseSettings init - MYSQL_HOST from env: {os.getenv('MYSQL_HOST', 'NOT SET')}")
|
|
logger.info(f"DatabaseSettings init - MYSQL_PORT from env: {os.getenv('MYSQL_PORT', 'NOT SET')}")
|
|
logger.info(f"DatabaseSettings init - MYSQL_DB from env: {os.getenv('MYSQL_DB', 'NOT SET')}")
|
|
logger.info(f"DatabaseSettings init - MYSQL_USER from env: {os.getenv('MYSQL_USER', 'NOT SET')}")
|
|
super().__init__(**kwargs)
|
|
logger.info(f"DatabaseSettings after init - host: {self.host}, port: {self.port}, database: {self.database}")
|
|
|
|
@property
|
|
def dsn(self) -> str:
|
|
"""Generate database connection string"""
|
|
return (
|
|
f"mysql+pymysql://{self.username}:{self.password}@"
|
|
f"{self.host}:{self.port}/{self.database}?charset=utf8mb4"
|
|
)
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class SecuritySettings(BaseSettings):
|
|
"""Security-related settings"""
|
|
|
|
master_key: str = Field(default="", env="MASTER_KEY")
|
|
jwt_secret_key: Optional[str] = Field(default=None, env="JWT_SECRET_KEY")
|
|
jwt_algorithm: str = Field(default="HS256", env="JWT_ALGORITHM")
|
|
access_token_expire_minutes: int = Field(default=30, env="ACCESS_TOKEN_EXPIRE_MINUTES")
|
|
refresh_token_expire_days: int = Field(default=7, env="REFRESH_TOKEN_EXPIRE_DAYS")
|
|
api_key_header: str = Field(default="X-API-Key", env="API_KEY_HEADER")
|
|
cors_origins: List[str] = Field(default=["*"], env="CORS_ORIGINS")
|
|
cors_credentials: bool = Field(default=True, env="CORS_CREDENTIALS")
|
|
cors_methods: List[str] = Field(default=["*"], env="CORS_METHODS")
|
|
cors_headers: List[str] = Field(default=["*"], env="CORS_HEADERS")
|
|
encryption_key: str = Field(default="", env="ENCRYPTION_KEY")
|
|
|
|
@field_validator("cors_origins", "cors_methods", "cors_headers", mode="before")
|
|
@classmethod
|
|
def parse_cors_list(cls, v):
|
|
if isinstance(v, str):
|
|
return [item.strip() for item in v.split(",")]
|
|
return v
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class OIDCSettings(BaseSettings):
|
|
"""OIDC/OAuth2 settings for Nextcloud integration"""
|
|
|
|
enabled: bool = Field(default=False, env="OIDC_ENABLED")
|
|
issuer: str = Field(default="", env="OIDC_ISSUER") # e.g., https://nextcloud.example.com
|
|
client_id: str = Field(default="", env="OIDC_CLIENT_ID")
|
|
client_secret: str = Field(default="", env="OIDC_CLIENT_SECRET")
|
|
redirect_uri: str = Field(default="", env="OIDC_REDIRECT_URI")
|
|
scope: str = Field(default="openid profile email groups", env="OIDC_SCOPE")
|
|
|
|
# Role mapping from OIDC groups
|
|
admin_groups: List[str] = Field(default=[], env="OIDC_ADMIN_GROUPS")
|
|
budget_reviewer_groups: List[str] = Field(default=[], env="OIDC_BUDGET_REVIEWER_GROUPS")
|
|
finance_reviewer_groups: List[str] = Field(default=[], env="OIDC_FINANCE_REVIEWER_GROUPS")
|
|
asta_groups: List[str] = Field(default=[], env="OIDC_ASTA_GROUPS")
|
|
|
|
# Auto-create users from OIDC
|
|
auto_create_users: bool = Field(default=True, env="OIDC_AUTO_CREATE_USERS")
|
|
|
|
@field_validator("admin_groups", "budget_reviewer_groups", "finance_reviewer_groups", "asta_groups", mode="before")
|
|
@classmethod
|
|
def parse_groups(cls, v):
|
|
if isinstance(v, str):
|
|
return [item.strip() for item in v.split(",") if item.strip()]
|
|
return v
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class EmailSettings(BaseSettings):
|
|
"""Email configuration settings"""
|
|
|
|
enabled: bool = Field(default=False, env="EMAIL_ENABLED")
|
|
smtp_host: str = Field(default="localhost", env="SMTP_HOST")
|
|
smtp_port: int = Field(default=587, env="SMTP_PORT")
|
|
smtp_tls: bool = Field(default=True, env="SMTP_TLS")
|
|
smtp_ssl: bool = Field(default=False, env="SMTP_SSL")
|
|
smtp_username: str = Field(default="", env="SMTP_USERNAME")
|
|
smtp_password: str = Field(default="", env="SMTP_PASSWORD")
|
|
from_email: str = Field(default="noreply@example.com", env="EMAIL_FROM")
|
|
from_name: str = Field(default="STUPA System", env="EMAIL_FROM_NAME")
|
|
|
|
# Email templates
|
|
verification_subject: str = Field(default="Verify your email", env="EMAIL_VERIFICATION_SUBJECT")
|
|
magic_link_subject: str = Field(default="Login to STUPA", env="EMAIL_MAGIC_LINK_SUBJECT")
|
|
application_notification_subject: str = Field(default="Application Status Update", env="EMAIL_APP_NOTIFICATION_SUBJECT")
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class RateLimitSettings(BaseSettings):
|
|
"""Rate limiting settings"""
|
|
|
|
enabled: bool = Field(default=True, env="RATE_LIMIT_ENABLED")
|
|
ip_per_minute: int = Field(default=60, env="RATE_IP_PER_MIN")
|
|
key_per_minute: int = Field(default=30, env="RATE_KEY_PER_MIN")
|
|
global_per_minute: int = Field(default=1000, env="RATE_GLOBAL_PER_MIN")
|
|
burst_size: int = Field(default=10, env="RATE_BURST_SIZE")
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class StorageSettings(BaseSettings):
|
|
"""File storage settings"""
|
|
|
|
upload_dir: Path = Field(default=Path("/tmp/uploads"), env="UPLOAD_DIR")
|
|
template_dir: Path = Field(default=Path("/tmp/templates"), env="TEMPLATE_DIR")
|
|
max_file_size: int = Field(default=10 * 1024 * 1024, env="MAX_FILE_SIZE") # 10MB
|
|
allowed_extensions: List[str] = Field(
|
|
default=["pdf", "json", "jpg", "jpeg", "png"],
|
|
env="ALLOWED_EXTENSIONS"
|
|
)
|
|
temp_dir: Path = Field(default=Path("/tmp"), env="TEMP_DIR")
|
|
attachment_storage: str = Field(default="database", env="ATTACHMENT_STORAGE") # database or filesystem
|
|
filesystem_path: Optional[Path] = Field(default=None, env="FILESYSTEM_PATH")
|
|
|
|
@field_validator("allowed_extensions", mode="before")
|
|
@classmethod
|
|
def parse_extensions(cls, v):
|
|
if isinstance(v, str):
|
|
return [ext.strip().lower() for ext in v.split(",")]
|
|
return v
|
|
|
|
def validate_paths(self):
|
|
"""Create directories if they don't exist"""
|
|
self.upload_dir.mkdir(parents=True, exist_ok=True)
|
|
self.template_dir.mkdir(parents=True, exist_ok=True)
|
|
self.temp_dir.mkdir(parents=True, exist_ok=True)
|
|
if self.filesystem_path:
|
|
self.filesystem_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class WorkflowSettings(BaseSettings):
|
|
"""Workflow configuration settings"""
|
|
|
|
# Required number of votes for approval
|
|
required_votes: int = Field(default=5, env="WORKFLOW_REQUIRED_VOTES")
|
|
|
|
# Vote thresholds (percentage)
|
|
approval_threshold: float = Field(default=50.0, env="WORKFLOW_APPROVAL_THRESHOLD")
|
|
|
|
# Timeouts (in days)
|
|
review_timeout_days: int = Field(default=14, env="WORKFLOW_REVIEW_TIMEOUT_DAYS")
|
|
voting_timeout_days: int = Field(default=7, env="WORKFLOW_VOTING_TIMEOUT_DAYS")
|
|
|
|
# Allow manual status changes
|
|
allow_manual_status_change: bool = Field(default=True, env="WORKFLOW_ALLOW_MANUAL_STATUS_CHANGE")
|
|
|
|
# Auto-lock on submission
|
|
auto_lock_on_submission: bool = Field(default=True, env="WORKFLOW_AUTO_LOCK_ON_SUBMISSION")
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class ApplicationSettings(BaseSettings):
|
|
"""General application settings"""
|
|
|
|
app_name: str = Field(default="STUPA PDF API", env="APP_NAME")
|
|
app_version: str = Field(default="3.0.0", env="APP_VERSION")
|
|
debug: bool = Field(default=False, env="DEBUG")
|
|
environment: str = Field(default="production", env="ENVIRONMENT")
|
|
log_level: str = Field(default="INFO", env="LOG_LEVEL")
|
|
timezone: str = Field(default="Europe/Berlin", env="TIMEZONE")
|
|
|
|
# Frontend URL for emails and redirects
|
|
frontend_url: str = Field(default="http://localhost:3001", env="FRONTEND_URL")
|
|
|
|
# API settings
|
|
api_prefix: str = Field(default="/api", env="API_PREFIX")
|
|
docs_url: Optional[str] = Field(default="/docs", env="DOCS_URL")
|
|
redoc_url: Optional[str] = Field(default="/redoc", env="REDOC_URL")
|
|
openapi_url: Optional[str] = Field(default="/openapi.json", env="OPENAPI_URL")
|
|
|
|
# Feature flags
|
|
enable_metrics: bool = Field(default=False, env="ENABLE_METRICS")
|
|
enable_tracing: bool = Field(default=False, env="ENABLE_TRACING")
|
|
enable_cache: bool = Field(default=True, env="ENABLE_CACHE")
|
|
cache_ttl: int = Field(default=300, env="CACHE_TTL") # seconds
|
|
|
|
# New feature flags
|
|
enable_form_designer: bool = Field(default=True, env="ENABLE_FORM_DESIGNER")
|
|
enable_pdf_upload: bool = Field(default=True, env="ENABLE_PDF_UPLOAD")
|
|
enable_workflow: bool = Field(default=True, env="ENABLE_WORKFLOW")
|
|
|
|
@field_validator("environment")
|
|
@classmethod
|
|
def validate_environment(cls, v):
|
|
allowed = ["development", "staging", "production", "testing"]
|
|
if v.lower() not in allowed:
|
|
raise ValueError(f"environment must be one of {allowed}")
|
|
return v.lower()
|
|
|
|
@field_validator("log_level")
|
|
@classmethod
|
|
def validate_log_level(cls, v):
|
|
allowed = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
|
if v.upper() not in allowed:
|
|
raise ValueError(f"log_level must be one of {allowed}")
|
|
return v.upper()
|
|
|
|
model_config = ConfigDict(env_prefix="")
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""Main settings class that aggregates all setting groups"""
|
|
|
|
database: DatabaseSettings = Field(default_factory=DatabaseSettings)
|
|
security: SecuritySettings = Field(default_factory=SecuritySettings)
|
|
oidc: OIDCSettings = Field(default_factory=OIDCSettings)
|
|
email: EmailSettings = Field(default_factory=EmailSettings)
|
|
rate_limit: RateLimitSettings = Field(default_factory=RateLimitSettings)
|
|
storage: StorageSettings = Field(default_factory=StorageSettings)
|
|
workflow: WorkflowSettings = Field(default_factory=WorkflowSettings)
|
|
app: ApplicationSettings = Field(default_factory=ApplicationSettings)
|
|
|
|
def __init__(self, **kwargs):
|
|
"""Initialize Settings with proper environment variable loading for nested models"""
|
|
super().__init__(**kwargs)
|
|
# Reinitialize nested settings to ensure they load environment variables
|
|
self.database = DatabaseSettings()
|
|
self.security = SecuritySettings()
|
|
self.oidc = OIDCSettings()
|
|
self.email = EmailSettings()
|
|
self.rate_limit = RateLimitSettings()
|
|
self.storage = StorageSettings()
|
|
self.workflow = WorkflowSettings()
|
|
self.app = ApplicationSettings()
|
|
|
|
# Dynamic configuration support
|
|
config_file: Optional[Path] = Field(default=None, env="CONFIG_FILE")
|
|
config_overrides: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
# Validate and create necessary directories
|
|
self.storage.validate_paths()
|
|
|
|
# Load configuration from file if specified
|
|
if self.config_file and self.config_file.exists():
|
|
self._load_config_file()
|
|
|
|
def _load_config_file(self):
|
|
"""Load configuration from JSON or YAML file"""
|
|
import json
|
|
try:
|
|
import yaml
|
|
has_yaml = True
|
|
except ImportError:
|
|
has_yaml = False
|
|
|
|
if self.config_file.suffix == ".json":
|
|
with open(self.config_file) as f:
|
|
config = json.load(f)
|
|
elif self.config_file.suffix in [".yaml", ".yml"] and has_yaml:
|
|
with open(self.config_file) as f:
|
|
config = yaml.safe_load(f)
|
|
else:
|
|
return
|
|
|
|
# Apply overrides from config file
|
|
self._apply_overrides(config)
|
|
|
|
def _apply_overrides(self, overrides: Dict[str, Any]):
|
|
"""Apply configuration overrides"""
|
|
for key, value in overrides.items():
|
|
if "." in key:
|
|
# Nested configuration
|
|
parts = key.split(".")
|
|
obj = self
|
|
for part in parts[:-1]:
|
|
obj = getattr(obj, part)
|
|
setattr(obj, parts[-1], value)
|
|
else:
|
|
if hasattr(self, key):
|
|
setattr(self, key, value)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Export settings as dictionary"""
|
|
return {
|
|
"database": {
|
|
"host": self.database.host,
|
|
"port": self.database.port,
|
|
"database": self.database.database,
|
|
},
|
|
"security": {
|
|
"cors_origins": self.security.cors_origins,
|
|
"api_key_header": self.security.api_key_header,
|
|
},
|
|
"oidc": {
|
|
"enabled": self.oidc.enabled,
|
|
"issuer": self.oidc.issuer,
|
|
},
|
|
"email": {
|
|
"enabled": self.email.enabled,
|
|
"from_email": self.email.from_email,
|
|
},
|
|
"rate_limit": {
|
|
"enabled": self.rate_limit.enabled,
|
|
"ip_per_minute": self.rate_limit.ip_per_minute,
|
|
},
|
|
"storage": {
|
|
"upload_dir": str(self.storage.upload_dir),
|
|
"template_dir": str(self.storage.template_dir),
|
|
"max_file_size": self.storage.max_file_size,
|
|
},
|
|
"workflow": {
|
|
"required_votes": self.workflow.required_votes,
|
|
"approval_threshold": self.workflow.approval_threshold,
|
|
},
|
|
"app": {
|
|
"name": self.app.app_name,
|
|
"version": self.app.app_version,
|
|
"environment": self.app.environment,
|
|
"enable_form_designer": self.app.enable_form_designer,
|
|
"enable_pdf_upload": self.app.enable_pdf_upload,
|
|
}
|
|
}
|
|
|
|
model_config = ConfigDict(
|
|
env_file=".env",
|
|
env_file_encoding="utf-8",
|
|
case_sensitive=False,
|
|
extra="allow"
|
|
)
|
|
|
|
|
|
@lru_cache()
|
|
def get_settings() -> Settings:
|
|
"""Get cached settings instance"""
|
|
return Settings()
|
|
|
|
|
|
# Convenience function for getting settings
|
|
settings = get_settings()
|