Samir Boulahtit
cad862f469
Replace direct User database model imports in API endpoints with UserContext schema, following the architecture principle that API routes should not import database models directly. Changes: - Create UserContext schema in models/schema/auth.py with from_user() factory - Update app/api/deps.py to return UserContext from all auth dependencies - Add _get_user_model() helper for functions needing User model access - Update 58 API endpoint files to use UserContext instead of User - Add noqa comments for 4 legitimate edge cases (enums, internal helpers) Architecture validation: 0 errors (down from 61), 11 warnings remain Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
139 lines
4.0 KiB
Python
139 lines
4.0 KiB
Python
# app/api/v1/admin/media.py
|
|
"""
|
|
Admin media management endpoints for vendor media libraries.
|
|
|
|
Allows admins to manage media files on behalf of vendors.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, File, Query, UploadFile
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_admin_api
|
|
from app.core.database import get_db
|
|
from app.services.media_service import media_service
|
|
from models.schema.auth import UserContext
|
|
from models.schema.media import (
|
|
MediaDetailResponse,
|
|
MediaItemResponse,
|
|
MediaListResponse,
|
|
MediaUploadResponse,
|
|
)
|
|
|
|
router = APIRouter(prefix="/media")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@router.get("/vendors/{vendor_id}", response_model=MediaListResponse)
|
|
def get_vendor_media_library(
|
|
vendor_id: int,
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
media_type: str | None = Query(None, description="image, video, document"),
|
|
folder: str | None = Query(None, description="Filter by folder"),
|
|
search: str | None = Query(None),
|
|
current_admin: UserContext = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Get media library for a specific vendor.
|
|
|
|
Admin can browse any vendor's media library.
|
|
"""
|
|
media_files, total = media_service.get_media_library(
|
|
db=db,
|
|
vendor_id=vendor_id,
|
|
skip=skip,
|
|
limit=limit,
|
|
media_type=media_type,
|
|
folder=folder,
|
|
search=search,
|
|
)
|
|
|
|
return MediaListResponse(
|
|
media=[MediaItemResponse.model_validate(m) for m in media_files],
|
|
total=total,
|
|
skip=skip,
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
@router.post("/vendors/{vendor_id}/upload", response_model=MediaUploadResponse)
|
|
async def upload_vendor_media(
|
|
vendor_id: int,
|
|
file: UploadFile = File(...),
|
|
folder: str | None = Query("products", description="products, general, etc."),
|
|
current_admin: UserContext = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Upload media file for a specific vendor.
|
|
|
|
Admin can upload media on behalf of any vendor.
|
|
Files are stored in vendor-specific directories.
|
|
"""
|
|
# Read file content
|
|
file_content = await file.read()
|
|
|
|
# Upload using service
|
|
media_file = await media_service.upload_file(
|
|
db=db,
|
|
vendor_id=vendor_id,
|
|
file_content=file_content,
|
|
filename=file.filename or "unnamed",
|
|
folder=folder or "products",
|
|
)
|
|
|
|
logger.info(f"Admin uploaded media for vendor {vendor_id}: {media_file.id}")
|
|
|
|
return MediaUploadResponse(
|
|
success=True,
|
|
message="File uploaded successfully",
|
|
media=MediaItemResponse.model_validate(media_file),
|
|
)
|
|
|
|
|
|
@router.get("/vendors/{vendor_id}/{media_id}", response_model=MediaDetailResponse)
|
|
def get_vendor_media_detail(
|
|
vendor_id: int,
|
|
media_id: int,
|
|
current_admin: UserContext = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Get detailed info about a specific media file.
|
|
"""
|
|
media_file = media_service.get_media_by_id(db=db, media_id=media_id)
|
|
|
|
# Verify media belongs to the specified vendor
|
|
if media_file.vendor_id != vendor_id:
|
|
from app.exceptions.media import MediaNotFoundException
|
|
raise MediaNotFoundException(media_id)
|
|
|
|
return MediaDetailResponse.model_validate(media_file)
|
|
|
|
|
|
@router.delete("/vendors/{vendor_id}/{media_id}")
|
|
def delete_vendor_media(
|
|
vendor_id: int,
|
|
media_id: int,
|
|
current_admin: UserContext = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Delete a media file for a vendor.
|
|
"""
|
|
media_file = media_service.get_media_by_id(db=db, media_id=media_id)
|
|
|
|
# Verify media belongs to the specified vendor
|
|
if media_file.vendor_id != vendor_id:
|
|
from app.exceptions.media import MediaNotFoundException
|
|
raise MediaNotFoundException(media_id)
|
|
|
|
media_service.delete_media(db=db, media_id=media_id)
|
|
|
|
logger.info(f"Admin deleted media {media_id} for vendor {vendor_id}")
|
|
|
|
return {"success": True, "message": "Media deleted successfully"}
|