Add file upload/attachments for items

backend/app/alembic/versions/a1b2c3d4e5f6_add_attachment_model.py

@@ -0,0 +1,58 @@
+"""Add attachment model
+
+Revision ID: a1b2c3d4e5f6
+Revises: fe56fa70289e
+Create Date: 2026-03-06 10:00:00.000000
+
+"""
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "a1b2c3d4e5f6"
+down_revision = "fe56fa70289e"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "attachment",
+        sa.Column(
+            "filename",
+            sqlmodel.sql.sqltypes.AutoString(length=255),
+            nullable=False,
+        ),
+        sa.Column(
+            "content_type",
+            sqlmodel.sql.sqltypes.AutoString(length=100),
+            nullable=False,
+        ),
+        sa.Column("size", sa.Integer(), nullable=False),
+        sa.Column(
+            "id", sa.Uuid(), nullable=False
+        ),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=True
+        ),
+        sa.Column(
+            "storage_path",
+            sqlmodel.sql.sqltypes.AutoString(length=512),
+            nullable=False,
+        ),
+        sa.Column("item_id", sa.Uuid(), nullable=False),
+        sa.Column("uploaded_by", sa.Uuid(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["item_id"], ["item.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["uploaded_by"], ["user.id"]
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
+
+def downgrade():
+    op.drop_table("attachment")

backend/app/api/main.py

@@ -1,13 +1,14 @@
 from fastapi import APIRouter
 
-from app.api.routes import items, login, private, users, utils
+from app.api.routes import attachments, items, login, private, users, utils
 from app.core.config import settings
 
 api_router = APIRouter()
 api_router.include_router(login.router)
 api_router.include_router(users.router)
 api_router.include_router(utils.router)
 api_router.include_router(items.router)
+api_router.include_router(attachments.router)
 
 
 if settings.ENVIRONMENT == "local":

backend/app/api/routes/attachments.py

@@ -0,0 +1,172 @@
+import os
+import re
+import uuid
+from pathlib import Path
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, UploadFile
+from fastapi.responses import FileResponse
+from sqlmodel import col, func, select
+
+from app.api.deps import CurrentUser, SessionDep
+from app.core.config import settings
+from app.models import (
+    Attachment,
+    AttachmentPublic,
+    AttachmentsPublic,
+    Item,
+    Message,
+)
+
+router = APIRouter(prefix="/items/{item_id}/attachments", tags=["attachments"])
+
+
+def _sanitize_filename(filename: str) -> str:
+    """Remove directory separators and other unsafe characters from filename."""
+    filename = os.path.basename(filename)
+    filename = re.sub(r"[^\w\s\-.]", "_", filename)
+    filename = filename.strip(". ")
+    if not filename:
+        filename = "unnamed"
+    return filename
+
+
+def _get_item_or_404(session: SessionDep, item_id: uuid.UUID) -> Item:
+    item = session.get(Item, item_id)
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return item
+
+
+def _check_item_access(item: Item, current_user: CurrentUser) -> None:
+    if not current_user.is_superuser and item.owner_id != current_user.id:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+
+
+@router.post("/", response_model=AttachmentPublic)
+def upload_attachment(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    item_id: uuid.UUID,
+    file: UploadFile,
+) -> Any:
+    """Upload a file attachment to an item."""
+    item = _get_item_or_404(session, item_id)
+    _check_item_access(item, current_user)
+
+    if file.content_type not in settings.ALLOWED_UPLOAD_TYPES:
+        raise HTTPException(
+            status_code=400,
+            detail=f"File type '{file.content_type}' is not allowed. "
+            f"Allowed types: {', '.join(settings.ALLOWED_UPLOAD_TYPES)}",
+        )
+
+    content = file.file.read()
+    if len(content) > settings.MAX_UPLOAD_SIZE:
+        raise HTTPException(
+            status_code=400,
+            detail=f"File size exceeds maximum of {settings.MAX_UPLOAD_SIZE} bytes",
+        )
+
+    safe_filename = _sanitize_filename(file.filename or "unnamed")
+    file_id = uuid.uuid4()
+    storage_filename = f"{file_id}_{safe_filename}"
+    upload_dir = Path(settings.UPLOAD_DIR) / str(item_id)
+    upload_dir.mkdir(parents=True, exist_ok=True)
+    storage_path = upload_dir / storage_filename
+
+    storage_path.write_bytes(content)
+
+    attachment = Attachment(
+        filename=safe_filename,
+        content_type=file.content_type or "application/octet-stream",
+        size=len(content),
+        storage_path=str(storage_path),
+        item_id=item_id,
+        uploaded_by=current_user.id,
+    )
+    session.add(attachment)
+    session.commit()
+    session.refresh(attachment)
+    return attachment
+
+
+@router.get("/", response_model=AttachmentsPublic)
+def list_attachments(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    item_id: uuid.UUID,
+) -> Any:
+    """List all attachments for an item."""
+    item = _get_item_or_404(session, item_id)
+    _check_item_access(item, current_user)
+
+    count_statement = (
+        select(func.count())
+        .select_from(Attachment)
+        .where(Attachment.item_id == item_id)
+    )
+    count = session.exec(count_statement).one()
+
+    statement = (
+        select(Attachment)
+        .where(Attachment.item_id == item_id)
+        .order_by(col(Attachment.created_at).desc())
+    )
+    attachments = session.exec(statement).all()
+
+    return AttachmentsPublic(data=attachments, count=count)
+
+
+@router.get("/{attachment_id}/download")
+def download_attachment(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    item_id: uuid.UUID,
+    attachment_id: uuid.UUID,
+) -> Any:
+    """Download an attachment file."""
+    item = _get_item_or_404(session, item_id)
+    _check_item_access(item, current_user)
+
+    attachment = session.get(Attachment, attachment_id)
+    if not attachment or attachment.item_id != item_id:
+        raise HTTPException(status_code=404, detail="Attachment not found")
+
+    file_path = Path(attachment.storage_path)
+    if not file_path.exists():
+        raise HTTPException(status_code=404, detail="Attachment file not found on disk")
+
+    return FileResponse(
+        path=str(file_path),
+        filename=attachment.filename,
+        media_type=attachment.content_type,
+    )
+
+
+@router.delete("/{attachment_id}")
+def delete_attachment(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    item_id: uuid.UUID,
+    attachment_id: uuid.UUID,
+) -> Message:
+    """Delete an attachment."""
+    item = _get_item_or_404(session, item_id)
+    _check_item_access(item, current_user)
+
+    attachment = session.get(Attachment, attachment_id)
+    if not attachment or attachment.item_id != item_id:
+        raise HTTPException(status_code=404, detail="Attachment not found")
+
+    file_path = Path(attachment.storage_path)
+    if file_path.exists():
+        file_path.unlink()
+
+    session.delete(attachment)
+    session.commit()
+    return Message(message="Attachment deleted successfully")

backend/app/core/config.py

@@ -83,6 +83,16 @@ def _set_default_emails_from(self) -> Self:
             self.EMAILS_FROM_NAME = self.PROJECT_NAME
         return self
 
+    UPLOAD_DIR: str = "uploads"
+    MAX_UPLOAD_SIZE: int = 10 * 1024 * 1024  # 10MB
+    ALLOWED_UPLOAD_TYPES: list[str] = [
+        "image/jpeg",
+        "image/png",
+        "image/gif",
+        "application/pdf",
+        "text/plain",
+    ]
+
     EMAIL_RESET_TOKEN_EXPIRE_HOURS: int = 48
 
     @computed_field  # type: ignore[prop-decorator]

backend/app/models.py

@@ -94,6 +94,9 @@ class Item(ItemBase, table=True):
         foreign_key="user.id", nullable=False, ondelete="CASCADE"
     )
     owner: User | None = Relationship(back_populates="items")
+    attachments: list["Attachment"] = Relationship(
+        back_populates="item", cascade_delete=True
+    )
 
 
 # Properties to return via API, id is always required
@@ -127,3 +130,36 @@ class TokenPayload(SQLModel):
 class NewPassword(SQLModel):
     token: str
     new_password: str = Field(min_length=8, max_length=128)
+
+
+# Attachment models
+class AttachmentBase(SQLModel):
+    filename: str = Field(max_length=255)
+    content_type: str = Field(max_length=100)
+    size: int
+
+
+class Attachment(AttachmentBase, table=True):
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    created_at: datetime | None = Field(
+        default_factory=get_datetime_utc,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+    storage_path: str = Field(max_length=512)
+    item_id: uuid.UUID = Field(
+        foreign_key="item.id", nullable=False, ondelete="CASCADE"
+    )
+    item: Item | None = Relationship(back_populates="attachments")
+    uploaded_by: uuid.UUID = Field(foreign_key="user.id", nullable=False)
+
+
+class AttachmentPublic(AttachmentBase):
+    id: uuid.UUID
+    item_id: uuid.UUID
+    uploaded_by: uuid.UUID
+    created_at: datetime | None = None
+
+
+class AttachmentsPublic(SQLModel):
+    data: list[AttachmentPublic]
+    count: int