Added full CRUD operations for spools

spoolson/api/v1/spool.py

@@ -1,7 +1,118 @@
 """Spool related endpoints."""
 
-from fastapi import APIRouter
+from datetime import datetime
+from typing import Annotated, Optional, Union
+
+from fastapi import APIRouter, Depends
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+from pydantic.error_wrappers import ErrorWrapper
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolson.api.v1.models import Message, Spool
+from spoolson.database import spool
+from spoolson.database.database import get_db_session
+from spoolson.exceptions import ItemCreateError
 
 router = APIRouter(
+    prefix="/spool",
     tags=["spool"],
 )
+
+# ruff: noqa: D103
+
+
+class SpoolParameters(BaseModel):
+    first_used: Optional[datetime] = Field(description="First logged occurence of spool usage.")
+    last_used: Optional[datetime] = Field(description="Last logged occurence of spool usage.")
+    filament_id: int = Field(description="The ID of the filament type of this spool.")
+    weight: Optional[float] = Field(
+        ge=0,
+        description=(
+            "Remaining weight of filament on the spool. "
+            "Leave empty to assume a full spool based on the weight parameters of the filament type."
+        ),
+    )
+    location: Optional[str] = Field(max_length=64, description="Where this spool can be found.")
+    lot_nr: Optional[str] = Field(max_length=64, description="Vendor manufacturing lot/batch number of the spool.")
+    comment: Optional[str] = Field(max_length=1024, description="Free text comment about this specific spool.")
+
+
+class SpoolUpdateParameters(SpoolParameters):
+    filament_id: Optional[int] = Field(description="The ID of the filament type of this spool.")
+
+
+@router.get("/")
+async def find(_filament: Union[int, None] = None) -> list[Spool]:
+    return []
+
+
+@router.get("/{spool_id}")
+async def get(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    spool_id: int,
+) -> Spool:
+    db_item = await spool.get_by_id(db, spool_id)
+    return Spool.from_db(db_item)
+
+
+@router.post(
+    "/",
+    response_model=Spool,
+    responses={
+        400: {"model": Message},
+    },
+)
+async def create(  # noqa: ANN201
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    body: SpoolParameters,
+):
+    try:
+        db_item = await spool.create(
+            db=db,
+            filament_id=body.filament_id,
+            weight=body.weight,
+            first_used=body.first_used,
+            last_used=body.last_used,
+            location=body.location,
+            lot_nr=body.lot_nr,
+            comment=body.comment,
+        )
+        return Spool.from_db(db_item)
+    except ItemCreateError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"message": str(exc)},
+        )
+
+
+@router.patch("/{spool_id}")
+async def update(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    spool_id: int,
+    body: SpoolUpdateParameters,
+) -> Spool:
+    patch_data = body.dict(exclude_unset=True)
+
+    if "filament_id" in patch_data and body.filament_id is None:
+        raise RequestValidationError(
+            [ErrorWrapper(ValueError("filament_id cannot be unset"), ("query", "filament_id"))],
+        )
+
+    db_item = await spool.update(
+        db=db,
+        spool_id=spool_id,
+        data=patch_data,
+    )
+
+    return Spool.from_db(db_item)
+
+
+@router.delete("/{spool_id}")
+async def delete(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    spool_id: int,
+) -> Message:
+    await spool.delete(db, spool_id)
+    return Message(message="Success!")

spoolson/database/models.py

@@ -33,8 +33,8 @@ class Filament(Base):
     price: Mapped[Optional[float]] = mapped_column()
     density: Mapped[float] = mapped_column()
     diameter: Mapped[float] = mapped_column()
-    weight: Mapped[Optional[float]] = mapped_column()
-    spool_weight: Mapped[Optional[float]] = mapped_column()
+    weight: Mapped[Optional[float]] = mapped_column(comment="The filament weight of a full spool (net weight).")
+    spool_weight: Mapped[Optional[float]] = mapped_column(comment="The weight of an empty spool.")
     article_number: Mapped[Optional[str]] = mapped_column(String(64))
     comment: Mapped[Optional[str]] = mapped_column(String(1024))
     # TODO: Print settings

spoolson/database/spool.py

@@ -0,0 +1,72 @@
+"""Helper functions for interacting with spool database objects."""
+
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolson.database import filament, models
+from spoolson.exceptions import ItemCreateError, ItemNotFoundError
+
+
+async def create(
+    *,
+    db: AsyncSession,
+    filament_id: int,
+    weight: Optional[float] = None,
+    first_used: Optional[datetime] = None,
+    last_used: Optional[datetime] = None,
+    location: Optional[str] = None,
+    lot_nr: Optional[str] = None,
+    comment: Optional[str] = None,
+) -> models.Spool:
+    """Add a new spool to the database. Leave weight empty to assume full spool."""
+    filament_item = await filament.get_by_id(db, filament_id)
+    if weight is None:
+        if filament_item.weight is None:
+            raise ItemCreateError("The weight for neither the spool nor its filament type is set.")
+        weight = filament_item.weight
+
+    db_item = models.Spool(
+        filament=filament_item,
+        weight=weight,
+        first_used=first_used,
+        last_used=last_used,
+        location=location,
+        lot_nr=lot_nr,
+        comment=comment,
+    )
+    db.add(db_item)
+    await db.flush()
+    return db_item
+
+
+async def get_by_id(db: AsyncSession, spool_id: int) -> models.Spool:
+    """Get a spool object from the database by the unique ID."""
+    spool = await db.get(models.Spool, spool_id)
+    if spool is None:
+        raise ItemNotFoundError(f"No spool with ID {spool_id} found.")
+    return spool
+
+
+async def update(
+    *,
+    db: AsyncSession,
+    spool_id: int,
+    data: dict,
+) -> models.Spool:
+    """Update the fields of a spool object."""
+    spool = await get_by_id(db, spool_id)
+    for k, v in data.items():
+        if k == "filament_id":
+            spool.filament = await filament.get_by_id(db, v)
+        else:
+            setattr(spool, k, v)
+    await db.flush()
+    return spool
+
+
+async def delete(db: AsyncSession, spool_id: int) -> None:
+    """Delete a spool object."""
+    spool = await get_by_id(db, spool_id)
+    await db.delete(spool)

spoolson/exceptions.py

@@ -3,3 +3,7 @@
 
 class ItemNotFoundError(Exception):
     pass
+
+
+class ItemCreateError(Exception):
+    pass