upload images to storage instead of DB and add image high quaiilty view
This commit is contained in:
165
app/routers/uploads.py
Normal file
165
app/routers/uploads.py
Normal file
@@ -0,0 +1,165 @@
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from fastapi import APIRouter, UploadFile, File, HTTPException, status, Depends
|
||||
from fastapi.responses import FileResponse
|
||||
from app.dependencies import get_current_user, get_optional_current_user
|
||||
from app.models.user import User
|
||||
|
||||
router = APIRouter(prefix="/api/uploads", tags=["uploads"])
|
||||
|
||||
# Storage directory for uploaded images
|
||||
STORAGE_DIR = Path("/home/shihaam/git/sarlink/mapmaker/storage/images")
|
||||
STORAGE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Allowed image types
|
||||
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}
|
||||
ALLOWED_MIME_TYPES = {
|
||||
"image/jpeg",
|
||||
"image/png",
|
||||
"image/gif",
|
||||
"image/webp",
|
||||
"image/bmp"
|
||||
}
|
||||
|
||||
# Maximum file size: 8MB
|
||||
MAX_FILE_SIZE = 8 * 1024 * 1024
|
||||
|
||||
|
||||
@router.post("/image")
|
||||
async def upload_image(
|
||||
file: UploadFile = File(...),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Upload an image file.
|
||||
Returns the file path that can be used to retrieve the image.
|
||||
"""
|
||||
# Validate file type by MIME type
|
||||
if file.content_type not in ALLOWED_MIME_TYPES:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Invalid file type. Allowed types: {', '.join(ALLOWED_MIME_TYPES)}"
|
||||
)
|
||||
|
||||
# Read file content
|
||||
content = await file.read()
|
||||
|
||||
# Validate file size
|
||||
if len(content) > MAX_FILE_SIZE:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"File too large. Maximum size is 8MB."
|
||||
)
|
||||
|
||||
# Get file extension
|
||||
original_extension = Path(file.filename).suffix.lower()
|
||||
if original_extension not in ALLOWED_EXTENSIONS:
|
||||
# Fallback to MIME type mapping
|
||||
mime_to_ext = {
|
||||
"image/jpeg": ".jpg",
|
||||
"image/png": ".png",
|
||||
"image/gif": ".gif",
|
||||
"image/webp": ".webp",
|
||||
"image/bmp": ".bmp"
|
||||
}
|
||||
original_extension = mime_to_ext.get(file.content_type, ".jpg")
|
||||
|
||||
# Generate random filename
|
||||
random_filename = f"{uuid.uuid4()}{original_extension}"
|
||||
file_path = STORAGE_DIR / random_filename
|
||||
|
||||
# Save file
|
||||
try:
|
||||
with open(file_path, "wb") as f:
|
||||
f.write(content)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to save file: {str(e)}"
|
||||
)
|
||||
|
||||
# Return the relative path
|
||||
return {
|
||||
"filename": random_filename,
|
||||
"path": f"/api/uploads/image/{random_filename}",
|
||||
"size": len(content)
|
||||
}
|
||||
|
||||
|
||||
@router.get("/image/{filename}")
|
||||
async def get_image(
|
||||
filename: str,
|
||||
current_user: Optional[User] = Depends(get_optional_current_user)
|
||||
):
|
||||
"""
|
||||
Retrieve an uploaded image.
|
||||
Public endpoint - requires either authenticated user or valid share token.
|
||||
"""
|
||||
# Validate filename to prevent directory traversal
|
||||
if ".." in filename or "/" in filename or "\\" in filename:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid filename"
|
||||
)
|
||||
|
||||
file_path = STORAGE_DIR / filename
|
||||
|
||||
# Check if file exists
|
||||
if not file_path.exists() or not file_path.is_file():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Image not found"
|
||||
)
|
||||
|
||||
# Determine media type based on extension
|
||||
extension = file_path.suffix.lower()
|
||||
media_type_map = {
|
||||
".jpg": "image/jpeg",
|
||||
".jpeg": "image/jpeg",
|
||||
".png": "image/png",
|
||||
".gif": "image/gif",
|
||||
".webp": "image/webp",
|
||||
".bmp": "image/bmp"
|
||||
}
|
||||
media_type = media_type_map.get(extension, "application/octet-stream")
|
||||
|
||||
return FileResponse(file_path, media_type=media_type)
|
||||
|
||||
|
||||
@router.delete("/image/{filename}")
|
||||
async def delete_image(
|
||||
filename: str,
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Delete an uploaded image.
|
||||
Requires authentication.
|
||||
"""
|
||||
# Validate filename to prevent directory traversal
|
||||
if ".." in filename or "/" in filename or "\\" in filename:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid filename"
|
||||
)
|
||||
|
||||
file_path = STORAGE_DIR / filename
|
||||
|
||||
# Check if file exists
|
||||
if not file_path.exists() or not file_path.is_file():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Image not found"
|
||||
)
|
||||
|
||||
# Delete file
|
||||
try:
|
||||
os.remove(file_path)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to delete file: {str(e)}"
|
||||
)
|
||||
|
||||
return {"message": "Image deleted successfully"}
|
||||
Reference in New Issue
Block a user