Spaces:
Sleeping
Sleeping
| from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File | |
| from datetime import datetime | |
| import os | |
| from app.core.config import settings as app_settings | |
| from sqlalchemy.orm import Session | |
| from typing import Any | |
| from app.db.base import get_db | |
| from app.models.settings import StoreSettings | |
| from app.schemas.settings import StoreSettingsResponse, StoreSettingsUpdate | |
| from app.api.auth import get_current_admin_user | |
| from app.schemas.api_response import APIResponse | |
| router = APIRouter(prefix="/settings", tags=["Settings"]) | |
| def get_settings(db: Session = Depends(get_db)) -> Any: | |
| settings = db.query(StoreSettings).first() | |
| if not settings: | |
| settings = StoreSettings() # create default | |
| db.add(settings) | |
| db.commit() | |
| db.refresh(settings) | |
| return APIResponse( | |
| isSuccess=True, | |
| value=StoreSettingsResponse.model_validate(settings).model_dump(), | |
| statusCode=200 | |
| ) | |
| def update_settings( | |
| settings_in: StoreSettingsUpdate, | |
| db: Session = Depends(get_db), | |
| current_admin=Depends(get_current_admin_user) | |
| ) -> Any: | |
| settings = db.query(StoreSettings).first() | |
| if not settings: | |
| settings = StoreSettings(**settings_in.model_dump(exclude_unset=True)) | |
| db.add(settings) | |
| else: | |
| update_data = settings_in.model_dump(exclude_unset=True) | |
| for field, value in update_data.items(): | |
| setattr(settings, field, value) | |
| db.commit() | |
| db.refresh(settings) | |
| return APIResponse( | |
| isSuccess=True, | |
| value=StoreSettingsResponse.model_validate(settings).model_dump(), | |
| statusCode=200 | |
| ) | |
| def upload_settings_image( | |
| file: UploadFile = File(...), | |
| db: Session = Depends(get_db), | |
| current_admin=Depends(get_current_admin_user) | |
| ) -> Any: | |
| # Convert file to base64 string | |
| file_bytes = file.file.read() | |
| import base64 | |
| ext = os.path.splitext(file.filename)[1].lower() | |
| mime_type = "image/jpeg" | |
| if ext in [".png"]: | |
| mime_type = "image/png" | |
| elif ext in [".gif"]: | |
| mime_type = "image/gif" | |
| elif ext in [".webp"]: | |
| mime_type = "image/webp" | |
| b64_encoded = base64.b64encode(file_bytes).decode('utf-8') | |
| data_uri = f"data:{mime_type};base64,{b64_encoded}" | |
| return APIResponse( | |
| isSuccess=True, | |
| value={"url": data_uri}, | |
| statusCode=200 | |
| ) | |