gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add settings management infrastructure: models, API, views, and database loading

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-07 22:31:06 +00:00
parent 24d09fc362
commit e1bb9de915
10 changed files with 1093 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/utils/config_loader.py
+138
View File
@@ -0,0 +1,138 @@
"""
Configuration loader that supports database-persisted settings with precedence.
This module provides functionality to:
- Load settings from database after app initialization
- Apply database settings over environment variables
- Dynamically reload settings when changed
"""
import logging
from typing import Any, Optional
from sqlalchemy.orm import Session
from app.models import ApplicationSettings
logger = logging.getLogger(__name__)
def load_settings_from_db(settings_obj, db_session: Session) -> None:
"""
Load settings from database and apply them to the settings object.
Database settings take precedence over environment variables and defaults.
This function should be called after database initialization.
Args:
settings_obj: The Settings instance to update
db_session: Database session to use for loading settings
"""
try:
db_settings = db_session.query(ApplicationSettings).all()
if not db_settings:
logger.info("No database settings found, using environment/defaults")
return
# Apply database settings to the settings object
updated_count = 0
for db_setting in db_settings:
key = db_setting.key
value = db_setting.value
# Check if the setting exists in the Settings class
if hasattr(settings_obj, key):
# Get the field info to determine the type
field_info = settings_obj.__fields__.get(key)
if field_info:
# Convert value to the appropriate type
converted_value = convert_setting_value(value, field_info.annotation)
# Set the attribute
setattr(settings_obj, key, converted_value)
updated_count += 1
logger.debug(f"Applied database setting: {key}")
if updated_count > 0:
logger.info(f"Loaded {updated_count} settings from database")
else:
logger.info("No applicable database settings found")
except Exception as e:
logger.error(f"Error loading settings from database: {e}")
# Don't fail application startup if database settings can't be loaded
logger.warning("Continuing with environment/default settings")
def convert_setting_value(value: Optional[str], field_type: Any) -> Any:
"""
Convert a string value from database to the appropriate type.
Args:
value: String value from database
field_type: Target type from Pydantic field annotation
Returns:
Converted value in the appropriate type
"""
if value is None:
return None
# Handle Optional types
origin = getattr(field_type, '__origin__', None)
if origin is Union:
# Get the non-None type from Union (for Optional)
args = getattr(field_type, '__args__', ())
field_type = next((arg for arg in args if arg is not type(None)), str)
# Convert based on type
if field_type == bool:
return value.lower() in ('true', '1', 'yes', 'y', 't')
elif field_type == int:
try:
return int(value)
except ValueError:
logger.warning(f"Failed to convert '{value}' to int, returning 0")
return 0
elif field_type == float:
try:
return float(value)
except ValueError:
logger.warning(f"Failed to convert '{value}' to float, returning 0.0")
return 0.0
elif field_type == list or getattr(field_type, '__origin__', None) == list:
# Handle list types - assume comma-separated values
if isinstance(value, str):
return [item.strip() for item in value.split(',') if item.strip()]
return value
else:
# Default to string
return str(value)
def reload_settings_from_db(settings_obj) -> bool:
"""
Reload settings from database.
This is useful after settings have been updated through the UI.
Note: Some settings require application restart to take effect.
Args:
settings_obj: The Settings instance to update
Returns:
True if reload was successful, False otherwise
"""
try:
from app.database import SessionLocal
db = SessionLocal()
try:
load_settings_from_db(settings_obj, db)
logger.info("Settings reloaded from database")
return True
finally:
db.close()
except Exception as e:
logger.error(f"Error reloading settings from database: {e}")
return False
app/utils/settings_service.py
+326
View File
@@ -0,0 +1,326 @@
"""
Service for managing application settings with database persistence.
This module provides functionality to:
- Load settings from database with precedence over environment variables
- Save settings to database
- Get setting metadata (descriptions, types, categories)
"""
import logging
from typing import Any, Dict, List, Optional
from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError
from app.models import ApplicationSettings
logger = logging.getLogger(__name__)
# Define setting metadata for UI display
SETTING_METADATA = {
# Core Settings
"database_url": {
"category": "Core",
"description": "Database connection URL (e.g., sqlite:///path/to/db.sqlite)",
"type": "string",
"sensitive": False,
"required": True,
"restart_required": True,
},
"redis_url": {
"category": "Core",
"description": "Redis connection URL for Celery task queue",
"type": "string",
"sensitive": False,
"required": True,
"restart_required": True,
},
"workdir": {
"category": "Core",
"description": "Working directory for file storage and processing",
"type": "string",
"sensitive": False,
"required": True,
"restart_required": True,
},
"external_hostname": {
"category": "Core",
"description": "External hostname for the application (e.g., docuelevate.example.com)",
"type": "string",
"sensitive": False,
"required": True,
"restart_required": True,
},
"debug": {
"category": "Core",
"description": "Enable debug mode for verbose logging",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": True,
},
"allow_file_delete": {
"category": "Core",
"description": "Allow deleting files from the database",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"gotenberg_url": {
"category": "Core",
"description": "Gotenberg service URL for document conversion",
"type": "string",
"sensitive": False,
"required": True,
"restart_required": True,
},
# Authentication Settings
"auth_enabled": {
"category": "Authentication",
"description": "Enable authentication for the application",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": True,
},
"session_secret": {
"category": "Authentication",
"description": "Secret key for session encryption (min 32 characters)",
"type": "string",
"sensitive": True,
"required": True,
"restart_required": True,
},
"admin_username": {
"category": "Authentication",
"description": "Admin username for local authentication",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": True,
},
"admin_password": {
"category": "Authentication",
"description": "Admin password for local authentication",
"type": "string",
"sensitive": True,
"required": False,
"restart_required": True,
},
# AI Services
"openai_api_key": {
"category": "AI Services",
"description": "OpenAI API key for metadata extraction",
"type": "string",
"sensitive": True,
"required": True,
"restart_required": False,
},
"openai_base_url": {
"category": "AI Services",
"description": "OpenAI API base URL (default: https://api.openai.com/v1)",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"openai_model": {
"category": "AI Services",
"description": "OpenAI model to use (e.g., gpt-4o-mini)",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"azure_ai_key": {
"category": "AI Services",
"description": "Azure AI key for document intelligence",
"type": "string",
"sensitive": True,
"required": True,
"restart_required": False,
},
"azure_region": {
"category": "AI Services",
"description": "Azure region for AI services",
"type": "string",
"sensitive": False,
"required": True,
"restart_required": False,
},
"azure_endpoint": {
"category": "AI Services",
"description": "Azure AI endpoint URL",
"type": "string",
"sensitive": False,
"required": True,
"restart_required": False,
},
# Add more settings metadata as needed...
}
def get_setting_from_db(db: Session, key: str) -> Optional[str]:
"""
Retrieve a setting value from the database.
Args:
db: Database session
key: Setting key to retrieve
Returns:
Setting value as string, or None if not found
"""
try:
setting = db.query(ApplicationSettings).filter(ApplicationSettings.key == key).first()
return setting.value if setting else None
except SQLAlchemyError as e:
logger.error(f"Error retrieving setting {key} from database: {e}")
return None
def save_setting_to_db(db: Session, key: str, value: Optional[str]) -> bool:
"""
Save or update a setting in the database.
Args:
db: Database session
key: Setting key
value: Setting value (as string)
Returns:
True if successful, False otherwise
"""
try:
setting = db.query(ApplicationSettings).filter(ApplicationSettings.key == key).first()
if setting:
setting.value = value
else:
setting = ApplicationSettings(key=key, value=value)
db.add(setting)
db.commit()
logger.info(f"Saved setting {key} to database")
return True
except SQLAlchemyError as e:
logger.error(f"Error saving setting {key} to database: {e}")
db.rollback()
return False
def get_all_settings_from_db(db: Session) -> Dict[str, str]:
"""
Retrieve all settings from the database.
Args:
db: Database session
Returns:
Dictionary of setting key-value pairs
"""
try:
settings = db.query(ApplicationSettings).all()
return {setting.key: setting.value for setting in settings}
except SQLAlchemyError as e:
logger.error(f"Error retrieving all settings from database: {e}")
return {}
def delete_setting_from_db(db: Session, key: str) -> bool:
"""
Delete a setting from the database.
Args:
db: Database session
key: Setting key to delete
Returns:
True if successful, False otherwise
"""
try:
setting = db.query(ApplicationSettings).filter(ApplicationSettings.key == key).first()
if setting:
db.delete(setting)
db.commit()
logger.info(f"Deleted setting {key} from database")
return True
return False
except SQLAlchemyError as e:
logger.error(f"Error deleting setting {key} from database: {e}")
db.rollback()
return False
def get_setting_metadata(key: str) -> Dict[str, Any]:
"""
Get metadata for a specific setting.
Args:
key: Setting key
Returns:
Dictionary containing setting metadata
"""
return SETTING_METADATA.get(key, {
"category": "Other",
"description": f"Setting: {key}",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
})
def get_settings_by_category() -> Dict[str, List[str]]:
"""
Get settings organized by category.
Returns:
Dictionary mapping category names to lists of setting keys
"""
categories = {}
for key, metadata in SETTING_METADATA.items():
category = metadata.get("category", "Other")
if category not in categories:
categories[category] = []
categories[category].append(key)
return categories
def validate_setting_value(key: str, value: str) -> tuple[bool, Optional[str]]:
"""
Validate a setting value based on its metadata.
Args:
key: Setting key
value: Setting value to validate
Returns:
Tuple of (is_valid, error_message)
"""
metadata = get_setting_metadata(key)
setting_type = metadata.get("type", "string")
# Check required fields
if metadata.get("required", False) and not value:
return False, f"{key} is required"
# Type-specific validation
if setting_type == "boolean":
if value.lower() not in ["true", "false", "1", "0", "yes", "no"]:
return False, f"{key} must be a boolean value (true/false)"
elif setting_type == "integer":
try:
int(value)
except ValueError:
return False, f"{key} must be an integer"
# Special validation for specific keys
if key == "session_secret" and value and len(value) < 32:
return False, "session_secret must be at least 32 characters"
return True, None