Renamed to Spoolman

Spoolson just didn't sound that good after all

spoolman/api/v1/filament.py

@@ -0,0 +1,111 @@
+"""Filament related endpoints."""
+
+from typing import Annotated, Optional, Union
+
+from fastapi import APIRouter, Depends
+from fastapi.exceptions import RequestValidationError
+from pydantic import BaseModel, Field
+from pydantic.error_wrappers import ErrorWrapper
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolman.api.v1.models import Filament, Message
+from spoolman.database import filament
+from spoolman.database.database import get_db_session
+
+router = APIRouter(
+    prefix="/filament",
+    tags=["filament"],
+)
+
+# ruff: noqa: D103
+
+
+class FilamentParameters(BaseModel):
+    name: Optional[str] = Field(
+        max_length=64,
+        description=(
+            "Filament name, to distinguish this filament type among others from the same vendor."
+            "Should contain its color for example."
+        ),
+    )
+    vendor_id: Optional[int] = Field(description="The ID of the vendor of this filament type.")
+    material: Optional[str] = Field(max_length=64, description="The material of this filament, e.g. PLA.")
+    price: Optional[float] = Field(ge=0, description="The price of this filament in the system configured currency.")
+    density: float = Field(gt=0, description="The density of this filament in g/cm3.")
+    diameter: float = Field(gt=0, description="The diameter of this filament in mm.")
+    weight: Optional[float] = Field(gt=0, description="The weight of the filament in a full spool.")
+    spool_weight: Optional[float] = Field(gt=0, description="The empty spool weight.")
+    article_number: Optional[str] = Field(max_length=64, description="Vendor article number, e.g. EAN, QR code, etc.")
+    comment: Optional[str] = Field(max_length=1024, description="Free text comment about this filament type.")
+
+
+class FilamentUpdateParameters(FilamentParameters):
+    density: Optional[float] = Field(gt=0, description="The density of this filament in g/cm3.")
+    diameter: Optional[float] = Field(gt=0, description="The diameter of this filament in mm.")
+
+
+@router.get("/")
+async def find(_vendor: Union[int, None] = None) -> list[Filament]:
+    return []
+
+
+@router.get("/{filament_id}")
+async def get(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    filament_id: int,
+) -> Filament:
+    db_item = await filament.get_by_id(db, filament_id)
+    return Filament.from_db(db_item)
+
+
+@router.post("/")
+async def create(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    body: FilamentParameters,
+) -> Filament:
+    db_item = await filament.create(
+        db=db,
+        density=body.density,
+        diameter=body.diameter,
+        name=body.name,
+        vendor_id=body.vendor_id,
+        material=body.material,
+        price=body.price,
+        weight=body.weight,
+        spool_weight=body.spool_weight,
+        article_number=body.article_number,
+        comment=body.comment,
+    )
+
+    return Filament.from_db(db_item)
+
+
+@router.patch("/{filament_id}")
+async def update(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    filament_id: int,
+    body: FilamentUpdateParameters,
+) -> Filament:
+    patch_data = body.dict(exclude_unset=True)
+
+    if "density" in patch_data and body.density is None:
+        raise RequestValidationError([ErrorWrapper(ValueError("density cannot be unset"), ("query", "density"))])
+    if "diameter" in patch_data and body.diameter is None:
+        raise RequestValidationError([ErrorWrapper(ValueError("diameter cannot be unset"), ("query", "diameter"))])
+
+    db_item = await filament.update(
+        db=db,
+        filament_id=filament_id,
+        data=patch_data,
+    )
+
+    return Filament.from_db(db_item)
+
+
+@router.delete("/{filament_id}")
+async def delete(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    filament_id: int,
+) -> Message:
+    await filament.delete(db, filament_id)
+    return Message(message="Success!")

spoolman/api/v1/models.py

@@ -0,0 +1,91 @@
+"""Pydantic data models for typing the FastAPI request/responses."""
+
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from spoolman.database import models
+
+
+class Message(BaseModel):
+    message: str = Field()
+
+
+class Vendor(BaseModel):
+    id: int = Field(description="Unique internal ID of this vendor.")
+    name: str = Field(max_length=64, description="Vendor name.")
+    comment: Optional[str] = Field(max_length=1024, description="Free text comment about this vendor.")
+
+    @staticmethod
+    def from_db(item: models.Vendor) -> "Vendor":
+        """Create a new Pydantic vendor object from a database vendor object."""
+        return Vendor(
+            id=item.id,
+            name=item.name,
+            comment=item.comment,
+        )
+
+
+class Filament(BaseModel):
+    id: int = Field(description="Unique internal ID of this filament type.")
+    name: Optional[str] = Field(
+        max_length=64,
+        description=(
+            "Filament name, to distinguish this filament type among others from the same vendor."
+            "Should contain its color for example."
+        ),
+    )
+    vendor: Optional[Vendor] = Field(description="The vendor of this filament type.")
+    material: Optional[str] = Field(max_length=64, description="The material of this filament, e.g. PLA.")
+    price: Optional[float] = Field(ge=0, description="The price of this filament in the system configured currency.")
+    density: float = Field(gt=0, description="The density of this filament in g/cm3.")
+    diameter: float = Field(gt=0, description="The diameter of this filament in mm.")
+    weight: Optional[float] = Field(gt=0, description="The weight of the filament in a full spool.")
+    spool_weight: Optional[float] = Field(gt=0, description="The empty spool weight.")
+    article_number: Optional[str] = Field(max_length=64, description="Vendor article number, e.g. EAN, QR code, etc.")
+    comment: Optional[str] = Field(max_length=1024, description="Free text comment about this filament type.")
+
+    @staticmethod
+    def from_db(item: models.Filament) -> "Filament":
+        """Create a new Pydantic filament object from a database filament object."""
+        return Filament(
+            id=item.id,
+            name=item.name,
+            vendor=Vendor.from_db(item.vendor) if item.vendor is not None else None,
+            material=item.material,
+            price=item.price,
+            density=item.density,
+            diameter=item.diameter,
+            weight=item.weight,
+            spool_weight=item.spool_weight,
+            article_number=item.article_number,
+            comment=item.comment,
+        )
+
+
+class Spool(BaseModel):
+    id: int = Field(description="Unique internal ID of this spool of filament.")
+    registered: datetime = Field(description="When the spool was registered in the database.")
+    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: Filament = Field(description="The filament type of this spool.")
+    weight: float = Field(ge=0, description="Remaining weight of filament on the spool.")
+    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.")
+
+    @staticmethod
+    def from_db(item: models.Spool) -> "Spool":
+        """Create a new Pydantic spool object from a database spool object."""
+        return Spool(
+            id=item.id,
+            registered=item.registered,
+            first_used=item.first_used,
+            last_used=item.last_used,
+            filament=Filament.from_db(item.filament),
+            weight=item.weight,
+            location=item.location,
+            lot_nr=item.lot_nr,
+            comment=item.comment,
+        )

spoolman/api/v1/router.py

@@ -0,0 +1,32 @@
+"""Router setup for the v1 version of the API."""
+
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+from starlette.requests import Request
+from starlette.responses import Response
+
+from spoolman.exceptions import ItemNotFoundError
+
+from . import filament, models, spool, vendor
+
+# ruff: noqa: D103
+
+app = FastAPI(
+    title="Spoolson REST API v1",
+    version="1.0.0",
+    root_path_in_servers=False,
+    responses={404: {"model": models.Message}},
+)
+
+
+@app.exception_handler(ItemNotFoundError)
+async def itemnotfounderror_exception_handler(_request: Request, exc: ItemNotFoundError) -> Response:
+    return JSONResponse(
+        status_code=404,
+        content={"message": str(exc)},
+    )
+
+
+app.include_router(filament.router)
+app.include_router(spool.router)
+app.include_router(vendor.router)

spoolman/api/v1/spool.py

@@ -0,0 +1,155 @@
+"""Spool related endpoints."""
+
+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 spoolman.api.v1.models import Message, Spool
+from spoolman.database import spool
+from spoolman.database.database import get_db_session
+from spoolman.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.")
+
+
+class SpoolUseParameters(BaseModel):
+    use_length: Optional[float] = Field(description="Length of filament to reduce by, in mm.")
+    use_weight: Optional[float] = Field(description="Filament weight to reduce by, in g.")
+
+
+@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!")
+
+
+@router.put(
+    "/{spool_id}/use",
+    response_model=Spool,
+    responses={
+        400: {"model": Message},
+    },
+)
+async def use(  # noqa: ANN201
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    spool_id: int,
+    body: SpoolUseParameters,
+):
+    if body.use_weight is not None and body.use_length is not None:
+        return JSONResponse(
+            status_code=400,
+            content={"message": "Only specify either use_weight or use_length."},
+        )
+
+    if body.use_weight is not None:
+        db_item = await spool.use_weight(db, spool_id, body.use_weight)
+        return Spool.from_db(db_item)
+
+    if body.use_length is not None:
+        db_item = await spool.use_length(db, spool_id, body.use_length)
+        return Spool.from_db(db_item)
+
+    return JSONResponse(
+        status_code=400,
+        content={"message": "Either use_weight or use_length must be specified."},
+    )

spoolman/api/v1/vendor.py

@@ -0,0 +1,87 @@
+"""Vendor related endpoints."""
+
+from typing import Annotated, Optional, Union
+
+from fastapi import APIRouter, Depends
+from fastapi.exceptions import RequestValidationError
+from pydantic import BaseModel, Field
+from pydantic.error_wrappers import ErrorWrapper
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolman.api.v1.models import Message, Vendor
+from spoolman.database import vendor
+from spoolman.database.database import get_db_session
+
+router = APIRouter(
+    prefix="/vendor",
+    tags=["vendor"],
+)
+
+# ruff: noqa: D103
+
+
+class VendorParameters(BaseModel):
+    name: str = Field(max_length=64, description="Vendor name.")
+    comment: Optional[str] = Field(max_length=1024, description="Free text comment about this vendor.")
+
+
+class VendorUpdateParameters(VendorParameters):
+    name: Optional[str] = Field(max_length=64, description="Vendor name.")
+    comment: Optional[str] = Field(max_length=1024, description="Free text comment about this vendor.")
+
+
+@router.get("/")
+async def find(_name: Union[int, None] = None) -> list[Vendor]:
+    return []
+
+
+@router.get("/{vendor_id}")
+async def get(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    vendor_id: int,
+) -> Vendor:
+    db_item = await vendor.get_by_id(db, vendor_id)
+    return Vendor.from_db(db_item)
+
+
+@router.post("/")
+async def create(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    body: VendorParameters,
+) -> Vendor:
+    db_item = await vendor.create(
+        db=db,
+        name=body.name,
+        comment=body.comment,
+    )
+
+    return Vendor.from_db(db_item)
+
+
+@router.patch("/{vendor_id}")
+async def update(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    vendor_id: int,
+    body: VendorUpdateParameters,
+) -> Vendor:
+    patch_data = body.dict(exclude_unset=True)
+
+    if "name" in patch_data and body.name is None:
+        raise RequestValidationError([ErrorWrapper(ValueError("name cannot be unset"), ("query", "name"))])
+
+    db_item = await vendor.update(
+        db=db,
+        vendor_id=vendor_id,
+        data=patch_data,
+    )
+
+    return Vendor.from_db(db_item)
+
+
+@router.delete("/{vendor_id}")
+async def delete(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    vendor_id: int,
+) -> Message:
+    await vendor.delete(db, vendor_id)
+    return Message(message="Success!")

spoolman/database/database.py

@@ -0,0 +1,54 @@
+"""SQLAlchemy database setup."""
+
+from collections.abc import AsyncGenerator
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
+
+from spoolman.database.models import Base
+
+
+class Database:
+    connection_url: str
+    engine: AsyncEngine
+    session_maker: async_sessionmaker[AsyncSession]
+
+    def __init__(self: "Database", connection_url: str) -> None:
+        """Construct the Database wrapper and set config parameters."""
+        self.connection_url = connection_url
+
+    def connect(self: "Database") -> None:
+        """Connect to the database."""
+        self.engine = create_async_engine(self.connection_url, connect_args={"check_same_thread": False})
+        self.session_maker = async_sessionmaker(self.engine, autocommit=False, autoflush=True)
+
+    async def create_tables(self: "Database") -> None:
+        """Create tables for all defined models."""
+        async with self.engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+
+
+__db: Optional[Database] = None
+
+
+async def setup_db(connection_url: str) -> None:
+    """Connect the singleton DB object."""
+    global __db  # noqa: PLW0603
+    __db = Database(connection_url)
+    __db.connect()
+    await __db.create_tables()
+
+
+async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
+    """Get a DB session to be used with FastAPI's dependency system."""
+    if __db is None:
+        raise RuntimeError("DB is not setup.")
+    async with __db.session_maker() as session:
+        try:
+            yield session
+            await session.commit()
+        except Exception as exc:
+            await session.rollback()
+            raise exc
+        finally:
+            await session.close()

spoolman/database/filament.py

@@ -0,0 +1,78 @@
+"""Helper functions for interacting with filament database objects."""
+
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolman.database import models, vendor
+from spoolman.exceptions import ItemNotFoundError
+
+
+async def create(
+    *,
+    db: AsyncSession,
+    density: float,
+    diameter: float,
+    name: Optional[str] = None,
+    vendor_id: Optional[int] = None,
+    material: Optional[str] = None,
+    price: Optional[float] = None,
+    weight: Optional[float] = None,
+    spool_weight: Optional[float] = None,
+    article_number: Optional[str] = None,
+    comment: Optional[str] = None,
+) -> models.Filament:
+    """Add a new filament to the database."""
+    vendor_item: Optional[models.Vendor] = None
+    if vendor_id is not None:
+        vendor_item = await vendor.get_by_id(db, vendor_id)
+
+    db_item = models.Filament(
+        name=name,
+        vendor=vendor_item,
+        material=material,
+        price=price,
+        density=density,
+        diameter=diameter,
+        weight=weight,
+        spool_weight=spool_weight,
+        article_number=article_number,
+        comment=comment,
+    )
+    db.add(db_item)
+    await db.flush()
+    return db_item
+
+
+async def get_by_id(db: AsyncSession, filament_id: int) -> models.Filament:
+    """Get a filament object from the database by the unique ID."""
+    filament = await db.get(models.Filament, filament_id)
+    if filament is None:
+        raise ItemNotFoundError(f"No filament with ID {filament_id} found.")
+    return filament
+
+
+async def update(
+    *,
+    db: AsyncSession,
+    filament_id: int,
+    data: dict,
+) -> models.Filament:
+    """Update the fields of a filament object."""
+    filament = await get_by_id(db, filament_id)
+    for k, v in data.items():
+        if k == "vendor_id":
+            if v is None:
+                filament.vendor = None
+            else:
+                filament.vendor = await vendor.get_by_id(db, v)
+        else:
+            setattr(filament, k, v)
+    await db.flush()
+    return filament
+
+
+async def delete(db: AsyncSession, filament_id: int) -> None:
+    """Delete a filament object."""
+    filament = await get_by_id(db, filament_id)
+    await db.delete(filament)

spoolman/database/models.py

@@ -0,0 +1,56 @@
+"""SQLAlchemy data models."""
+
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import ForeignKey, Integer, String
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+from sqlalchemy.sql import func
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class Vendor(Base):
+    __tablename__ = "vendor"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    name: Mapped[str] = mapped_column(String(64))
+    comment: Mapped[Optional[str]] = mapped_column(String(1024))
+    filaments: Mapped[list["Filament"]] = relationship(back_populates="vendor")
+
+
+class Filament(Base):
+    __tablename__ = "filament"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    name: Mapped[Optional[str]] = mapped_column(String(64))
+    vendor_id: Mapped[Optional[int]] = mapped_column(ForeignKey("vendor.id"))
+    vendor: Mapped[Optional["Vendor"]] = relationship(back_populates="filaments")
+    spools: Mapped[list["Spool"]] = relationship(back_populates="filament")
+    material: Mapped[Optional[str]] = mapped_column(String(64))
+    price: Mapped[Optional[float]] = mapped_column()
+    density: Mapped[float] = mapped_column()
+    diameter: Mapped[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
+    # TODO: Color?
+
+
+class Spool(Base):
+    __tablename__ = "spool"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    registered: Mapped[datetime] = mapped_column(default=func.now())
+    first_used: Mapped[Optional[datetime]] = mapped_column()
+    last_used: Mapped[Optional[datetime]] = mapped_column()
+    filament_id: Mapped[int] = mapped_column(ForeignKey("filament.id"))
+    filament: Mapped["Filament"] = relationship(back_populates="spools")
+    weight: Mapped[float] = mapped_column()
+    location: Mapped[Optional[str]] = mapped_column(String(64))
+    lot_nr: Mapped[Optional[str]] = mapped_column(String(64))
+    comment: Mapped[Optional[str]] = mapped_column(String(1024))

spoolman/database/spool.py

@@ -0,0 +1,120 @@
+"""Helper functions for interacting with spool database objects."""
+
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolman.database import filament, models
+from spoolman.exceptions import ItemCreateError, ItemNotFoundError
+from spoolman.math import weight_from_length
+
+
+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, with_for_update: Optional[bool] = None) -> models.Spool:
+    """Get a spool object from the database by the unique ID."""
+    spool = await db.get(models.Spool, spool_id, with_for_update=with_for_update)  # type: ignore  # noqa: PGH003
+    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)
+
+
+# TODO: Make unit tests for race conditions on these
+async def use_weight(db: AsyncSession, spool_id: int, weight: float) -> models.Spool:
+    """Reduce the weight of a spool.
+
+    Does nothing if the spool is empty.
+
+    Args:
+        db (AsyncSession): Database session
+        spool_id (int): Spool ID
+        weight (float): Weight loss in grams
+
+    Returns:
+        models.Spool: Updated spool object
+    """
+    spool = await get_by_id(db, spool_id, with_for_update=True)
+    spool.weight -= min(spool.weight, weight)
+    await db.flush()
+    return spool
+
+
+async def use_length(db: AsyncSession, spool_id: int, length: float) -> models.Spool:
+    """Reduce the weight of a spool by using a length of filament.
+
+    Does nothing if the spool is empty.
+
+    Args:
+        db (AsyncSession): Database session
+        spool_id (int): Spool ID
+        length (float): Length of filament to reduce by
+
+    Returns:
+        models.Spool: Updated spool object
+    """
+    spool = await get_by_id(db, spool_id, with_for_update=True)
+
+    filament = spool.filament
+
+    weight = weight_from_length(
+        length=length,
+        radius=filament.diameter / 2,
+        density=filament.density,
+    )
+    spool.weight -= min(spool.weight, weight)
+    await db.flush()
+    return spool

spoolman/database/vendor.py

@@ -0,0 +1,52 @@
+"""Helper functions for interacting with vendor database objects."""
+
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolman.database import models
+from spoolman.exceptions import ItemNotFoundError
+
+
+async def create(
+    *,
+    db: AsyncSession,
+    name: Optional[str] = None,
+    comment: Optional[str] = None,
+) -> models.Vendor:
+    """Add a new vendor to the database."""
+    db_item = models.Vendor(
+        name=name,
+        comment=comment,
+    )
+    db.add(db_item)
+    await db.flush()
+    return db_item
+
+
+async def get_by_id(db: AsyncSession, vendor_id: int) -> models.Vendor:
+    """Get a vendor object from the database by the unique ID."""
+    vendor = await db.get(models.Vendor, vendor_id)
+    if vendor is None:
+        raise ItemNotFoundError(f"No vendor with ID {vendor_id} found.")
+    return vendor
+
+
+async def update(
+    *,
+    db: AsyncSession,
+    vendor_id: int,
+    data: dict,
+) -> models.Vendor:
+    """Update the fields of a vendor object."""
+    vendor = await get_by_id(db, vendor_id)
+    for k, v in data.items():
+        setattr(vendor, k, v)
+    await db.flush()
+    return vendor
+
+
+async def delete(db: AsyncSession, vendor_id: int) -> None:
+    """Delete a vendor object."""
+    vendor = await get_by_id(db, vendor_id)
+    await db.delete(vendor)

spoolman/exceptions.py

@@ -0,0 +1,9 @@
+"""Various exceptions used."""
+
+
+class ItemNotFoundError(Exception):
+    pass
+
+
+class ItemCreateError(Exception):
+    pass

spoolman/main.py

@@ -0,0 +1,21 @@
+"""Main entrypoint to the server."""
+
+
+import uvicorn
+from fastapi import FastAPI
+
+from spoolman.api.v1.router import app as v1_app
+from spoolman.database.database import setup_db
+
+app = FastAPI(debug=True)
+app.mount("/api/v1", v1_app)
+
+
+@app.on_event("startup")
+async def startup() -> None:
+    """Run the service's startup sequence."""
+    await setup_db("sqlite+aiosqlite:///./sql_app.db")
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)

spoolman/math.py

@@ -0,0 +1,19 @@
+"""Various math-related functions."""
+
+import math
+
+
+def weight_from_length(*, length: float, radius: float, density: float) -> float:
+    """Calculate the weight of a piece of filament.
+
+    Args:
+        length (float): Filament length in mm
+        radius (float): Filament radius in mm
+        density (float): Density of filament material in g/cm3
+
+    Returns:
+        float: Weight in g
+    """
+    volume_mm3 = length * math.pi * radius * radius
+    volume_cm3 = volume_mm3 / 1000
+    return density * volume_cm3