"""System settings router - read/update settings, upload company logo.""" from pathlib import Path from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from config import settings from database import get_db from middleware.api_key import get_current_user, require_admin_user from models.setting import SystemSetting from models.user import User router = APIRouter(prefix="/api/settings", tags=["settings"]) @router.get("/") async def get_settings( user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Get all system settings as a dictionary. Returns key-value pairs of all settings. Available to all authenticated users. """ result = await db.execute(select(SystemSetting)) settings_list = result.scalars().all() settings_dict = { setting.setting_key: setting.setting_value for setting in settings_list } return settings_dict @router.put("/") async def update_settings( settings_data: dict[str, str], user: User = Depends(require_admin_user), db: AsyncSession = Depends(get_db), ): """Update multiple system settings. Expects a JSON object with setting_key: setting_value pairs. Only admins can update settings. """ updated_keys = [] for key, value in settings_data.items(): # Check if setting exists result = await db.execute( select(SystemSetting).where(SystemSetting.setting_key == key) ) setting = result.scalar_one_or_none() if setting is None: # Create new setting (assume type 'string' for new settings) setting = SystemSetting( setting_key=key, setting_value=value, setting_type="string", updated_by=user.id, ) db.add(setting) else: # Update existing setting setting.setting_value = value setting.updated_by = user.id updated_keys.append(key) await db.flush() return { "message": f"Updated {len(updated_keys)} settings", "updated_keys": updated_keys, } @router.post("/logo") async def upload_company_logo( file: UploadFile = File(...), user: User = Depends(require_admin_user), db: AsyncSession = Depends(get_db), ): """Upload company logo. Saves logo to uploads/logos/ and updates company_logo_path setting. Only admins can upload the logo. """ # Validate file type (must be image) allowed_types = {"image/jpeg", "image/png", "image/gif", "image/webp", "image/svg+xml"} if file.content_type not in allowed_types: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"File type {file.content_type} not allowed. Must be an image.", ) # Read file content content = await file.read() file_size = len(content) # Validate file size max_bytes = settings.max_upload_size_mb * 1024 * 1024 if file_size > max_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"File size {file_size} bytes exceeds maximum {settings.max_upload_size_mb}MB", ) # Create logos directory logos_dir = settings.upload_path / "logos" logos_dir.mkdir(parents=True, exist_ok=True) # Use a fixed filename for the company logo file_extension = Path(file.filename).suffix logo_filename = f"company_logo{file_extension}" logo_path = logos_dir / logo_filename # Write file logo_path.write_bytes(content) # Update setting in database relative_path = logo_path.relative_to(settings.upload_path) logo_path_str = str(relative_path).replace("\\", "/") result = await db.execute( select(SystemSetting).where(SystemSetting.setting_key == "company_logo_path") ) setting = result.scalar_one_or_none() if setting is None: setting = SystemSetting( setting_key="company_logo_path", setting_value=logo_path_str, setting_type="string", description="Path to company logo file", updated_by=user.id, ) db.add(setting) else: setting.setting_value = logo_path_str setting.updated_by = user.id await db.flush() return { "message": "Company logo uploaded successfully", "logo_path": logo_path_str, "file_size": file_size, }