diff --git a/spoolson/api/v1/spool.py b/spoolson/api/v1/spool.py index 0248feb..3177b9e 100644 --- a/spoolson/api/v1/spool.py +++ b/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!") diff --git a/spoolson/database/models.py b/spoolson/database/models.py index 01c6af9..1090261 100644 --- a/spoolson/database/models.py +++ b/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 diff --git a/spoolson/database/spool.py b/spoolson/database/spool.py new file mode 100644 index 0000000..c0fbfe5 --- /dev/null +++ b/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) diff --git a/spoolson/exceptions.py b/spoolson/exceptions.py index ff06bb3..97baeda 100644 --- a/spoolson/exceptions.py +++ b/spoolson/exceptions.py @@ -3,3 +3,7 @@ class ItemNotFoundError(Exception): pass + + +class ItemCreateError(Exception): + pass