allucanget/ai.allucanget.biz
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement image upload functionality with metadata storage; update frontend to display uploaded images

Browse Source 
Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
zwitschi
2026-04-29 13:10:20 +02:00
parent 20de65ad01
commit e1d74fe163
10 changed files with 464 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/db.py
+11
View File
@@ -64,3 +64,14 @@ def _run_migrations(conn: duckdb.DuckDBPyConnection) -> None:
revoked BOOLEAN DEFAULT false
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS uploaded_images (
id UUID DEFAULT uuid() PRIMARY KEY,
user_id UUID NOT NULL,
filename VARCHAR NOT NULL,
content_type VARCHAR NOT NULL,
file_path VARCHAR NOT NULL,
size_bytes BIGINT NOT NULL,
created_at TIMESTAMP DEFAULT now()
)
""")
backend/app/main.py
+2
View File
@@ -3,6 +3,7 @@ from .routers import users as users_router
from .routers import admin as admin_router
from .routers import ai as ai_router
from .routers import generate as generate_router
from .routers import images as images_router
from .db import close_db, init_db
import os
from contextlib import asynccontextmanager
@@ -41,6 +42,7 @@ app.include_router(users_router.router)
app.include_router(admin_router.router)
app.include_router(ai_router.router)
app.include_router(generate_router.router)
app.include_router(images_router.router)
@app.get("/health", tags=["health"])
backend/app/routers/images.py
+117
View File
@@ -0,0 +1,117 @@
"""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}/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])
backend/requirements.txt
+1
View File
@@ -17,4 +17,5 @@ passlib==1.7.4
pydantic
python-dotenv
python-jose
python-multipart
uvicorn
backend/tests/test_images.py
+184
View File
@@ -0,0 +1,184 @@
"""Tests for image upload and retrieval endpoints."""
import io
import os
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app
from app import db as db_module
os.environ.setdefault("JWT_SECRET", "test-secret-key-for-testing-only")
# Use a temp dir so file I/O works without polluting project data/
os.environ.setdefault("UPLOAD_DIR", "/tmp/test_uploads")
@pytest.fixture(autouse=True)
def fresh_db():
db_module._conn = None
db_module.init_db(":memory:")
yield
db_module.close_db()
db_module._conn = None
@pytest_asyncio.fixture
async def client(fresh_db):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
async def _user_token(client) -> str:
await client.post("/auth/register", json={"email": "user@example.com", "password": "secret123"})
resp = await client.post("/auth/login", json={"email": "user@example.com", "password": "secret123"})
return resp.json()["access_token"]
async def _other_token(client) -> str:
await client.post("/auth/register", json={"email": "other@example.com", "password": "secret123"})
resp = await client.post("/auth/login", json={"email": "other@example.com", "password": "secret123"})
return resp.json()["access_token"]
def _png_bytes() -> bytes:
"""Minimal valid 1x1 PNG."""
return (
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
b"\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f\x00"
b"\x00\x01\x01\x00\x05\x18\xd4n\x00\x00\x00\x00IEND\xaeB`\x82"
)
# ---------------------------------------------------------------------------
# POST /images/upload
# ---------------------------------------------------------------------------
async def test_upload_image_success(client):
token = await _user_token(client)
resp = await client.post(
"/images/upload",
files={"file": ("test.png", io.BytesIO(_png_bytes()), "image/png")},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 201
data = resp.json()
assert data["filename"] == "test.png"
assert data["content_type"] == "image/png"
assert "id" in data
assert data["size_bytes"] > 0
async def test_upload_image_unauthenticated(client):
resp = await client.post(
"/images/upload",
files={"file": ("test.png", io.BytesIO(_png_bytes()), "image/png")},
)
assert resp.status_code == 401
async def test_upload_image_unsupported_type(client):
token = await _user_token(client)
resp = await client.post(
"/images/upload",
files={"file": ("doc.pdf", io.BytesIO(
b"%PDF-fake"), "application/pdf")},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 415
async def test_upload_image_too_large(client, monkeypatch):
import app.routers.images as images_mod
monkeypatch.setattr(images_mod, "MAX_SIZE_BYTES", 5)
token = await _user_token(client)
resp = await client.post(
"/images/upload",
files={"file": ("big.png", io.BytesIO(b"x" * 10), "image/png")},
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 413
# ---------------------------------------------------------------------------
# GET /images/
# ---------------------------------------------------------------------------
async def test_list_images_empty(client):
token = await _user_token(client)
resp = await client.get("/images/", headers={"Authorization": f"Bearer {token}"})
assert resp.status_code == 200
assert resp.json() == []
async def test_list_images_returns_own_only(client):
token = await _user_token(client)
other = await _other_token(client)
# Upload one image as user, one as other
for tok, name in [(token, "mine.png"), (other, "theirs.png")]:
await client.post(
"/images/upload",
files={"file": (name, io.BytesIO(_png_bytes()), "image/png")},
headers={"Authorization": f"Bearer {tok}"},
)
resp = await client.get("/images/", headers={"Authorization": f"Bearer {token}"})
assert resp.status_code == 200
items = resp.json()
assert len(items) == 1
assert items[0]["filename"] == "mine.png"
async def test_list_images_unauthenticated(client):
resp = await client.get("/images/")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# GET /images/{id}/file
# ---------------------------------------------------------------------------
async def test_serve_image_success(client):
token = await _user_token(client)
up = await client.post(
"/images/upload",
files={"file": ("pixel.png", io.BytesIO(_png_bytes()), "image/png")},
headers={"Authorization": f"Bearer {token}"},
)
image_id = up.json()["id"]
resp = await client.get(
f"/images/{image_id}/file",
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("image/png")
assert resp.content == _png_bytes()
async def test_serve_image_wrong_user(client):
token = await _user_token(client)
other = await _other_token(client)
up = await client.post(
"/images/upload",
files={"file": ("secret.png", io.BytesIO(_png_bytes()), "image/png")},
headers={"Authorization": f"Bearer {token}"},
)
image_id = up.json()["id"]
resp = await client.get(
f"/images/{image_id}/file",
headers={"Authorization": f"Bearer {other}"},
)
assert resp.status_code == 403
async def test_serve_image_not_found(client):
token = await _user_token(client)
resp = await client.get(
"/images/00000000-0000-0000-0000-000000000000/file",
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 404