151 lines
4.7 KiB
Python
151 lines
4.7 KiB
Python
"""Images router: upload reference images and list user's uploads."""
|
|
import os
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, UploadFile, status
|
|
from fastapi.responses import FileResponse
|
|
|
|
from ..db import get_conn, get_write_lock
|
|
from ..dependencies import get_current_user
|
|
|
|
router = APIRouter(prefix="/images", tags=["images"])
|
|
|
|
UPLOAD_DIR = os.getenv("UPLOAD_DIR", "data/uploads")
|
|
MAX_SIZE_BYTES = 10 * 1024 * 1024 # 10 MB
|
|
ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
|
|
|
|
|
|
@router.post("/upload", status_code=status.HTTP_201_CREATED)
|
|
async def upload_image(
|
|
file: UploadFile,
|
|
current_user: dict = Depends(get_current_user),
|
|
) -> dict:
|
|
"""Upload a reference image and store metadata in DuckDB."""
|
|
if file.content_type not in ALLOWED_CONTENT_TYPES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
|
detail=f"Unsupported content type '{file.content_type}'. Allowed: {sorted(ALLOWED_CONTENT_TYPES)}",
|
|
)
|
|
|
|
data = await file.read()
|
|
if len(data) > MAX_SIZE_BYTES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail=f"File exceeds maximum allowed size of {MAX_SIZE_BYTES // (1024*1024)} MB.",
|
|
)
|
|
|
|
user_id = current_user["id"]
|
|
image_id = str(uuid.uuid4())
|
|
ext = (file.filename or "").rsplit(
|
|
".", 1)[-1].lower() if "." in (file.filename or "") else "bin"
|
|
safe_filename = f"{image_id}.{ext}"
|
|
user_dir = os.path.join(UPLOAD_DIR, user_id)
|
|
os.makedirs(user_dir, exist_ok=True)
|
|
file_path = os.path.join(user_dir, safe_filename)
|
|
|
|
with open(file_path, "wb") as f:
|
|
f.write(data)
|
|
|
|
async with get_write_lock():
|
|
conn = get_conn()
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO uploaded_images (id, user_id, filename, content_type, file_path, size_bytes)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
[image_id, user_id, file.filename or safe_filename,
|
|
file.content_type, file_path, len(data)],
|
|
)
|
|
|
|
return {
|
|
"id": image_id,
|
|
"filename": file.filename or safe_filename,
|
|
"content_type": file.content_type,
|
|
"size_bytes": len(data),
|
|
}
|
|
|
|
|
|
@router.get("/", status_code=status.HTTP_200_OK)
|
|
async def list_images(
|
|
current_user: dict = Depends(get_current_user),
|
|
) -> list[dict]:
|
|
"""Return all uploaded images for the current user."""
|
|
conn = get_conn()
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, filename, content_type, size_bytes, created_at
|
|
FROM uploaded_images
|
|
WHERE user_id = ?
|
|
ORDER BY created_at DESC
|
|
""",
|
|
[current_user["id"]],
|
|
).fetchall()
|
|
|
|
return [
|
|
{
|
|
"id": str(row[0]),
|
|
"filename": row[1],
|
|
"content_type": row[2],
|
|
"size_bytes": row[3],
|
|
"created_at": row[4].isoformat() if row[4] else None,
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
|
|
@router.get("/{image_id}", status_code=status.HTTP_200_OK)
|
|
async def get_image_details(
|
|
image_id: str,
|
|
current_user: dict = Depends(get_current_user),
|
|
) -> dict:
|
|
"""Return metadata for a single uploaded image."""
|
|
conn = get_conn()
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, filename, content_type, size_bytes, created_at
|
|
FROM uploaded_images
|
|
WHERE id = ? AND user_id = ?
|
|
""",
|
|
[image_id, current_user["id"]],
|
|
).fetchone()
|
|
|
|
if not row:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Image not found"
|
|
)
|
|
|
|
return {
|
|
"id": str(row[0]),
|
|
"filename": row[1],
|
|
"content_type": row[2],
|
|
"size_bytes": row[3],
|
|
"created_at": row[4].isoformat() if row[4] else None,
|
|
}
|
|
|
|
|
|
@router.get("/{image_id}/file", status_code=status.HTTP_200_OK)
|
|
async def serve_image(
|
|
image_id: str,
|
|
current_user: dict = Depends(get_current_user),
|
|
) -> FileResponse:
|
|
"""Serve the raw image file. Only accessible by the owning user."""
|
|
conn = get_conn()
|
|
row = conn.execute(
|
|
"SELECT file_path, content_type, user_id FROM uploaded_images WHERE id = ?",
|
|
[image_id],
|
|
).fetchone()
|
|
|
|
if row is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Image not found.")
|
|
if str(row[2]) != current_user["id"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Access denied.")
|
|
|
|
file_path: str = row[0]
|
|
if not os.path.isfile(file_path):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Image file missing.")
|
|
|
|
return FileResponse(file_path, media_type=row[1])
|