Filaments can now be created

pyproject.toml

@@ -8,7 +8,7 @@ license = {text = "AGPL-3.0-only"}
 
 [tool.ruff]
 select = ["ALL"]
-ignore = ["A003", "D101", "D104", "S104"]
+ignore = ["A003", "D101", "D104", "S104", "TRY201", "TRY003", "EM101", "EM102"]
 line-length = 120
 target-version = "py39"
 

requirements.txt

@@ -2,3 +2,5 @@ uvicorn[standard]==0.21.1
 fastapi==0.95.0
 SQLAlchemy[asyncio]==2.0.8
 pydantic==1.10.7
+
+aiosqlite==0.18.0

spoolson/api/v1/filament.py

@@ -1,9 +1,14 @@
 """Filament related endpoints."""
 
-from typing import Union
+from typing import Annotated, Optional, Union
-from fastapi import APIRouter
+
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel, Field
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from spoolson.api.v1.models import Filament, Vendor
+from spoolson.database.database import get_db_session
+from spoolson.database.filament import create_filament
 
 router = APIRouter(
     prefix="/filament",
@@ -13,13 +18,32 @@ router = APIRouter(
 # 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.")
+
+
 @router.get("/")
-async def find(vendor: Union[int, None] = None) -> list[Filament]:
+async def find(_vendor: Union[int, None] = None) -> list[Filament]:
     return []
 
 
 @router.get("/{filament_id}")
-async def get(filament_id: int) -> Filament:
+async def get(_filament_id: int) -> Filament:
     return Filament(
         id=0,
         name=None,
@@ -40,25 +64,26 @@ async def get(filament_id: int) -> Filament:
 
 
 @router.post("/")
-async def create() -> Filament:
+async def create(
-    return Filament(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
-        id=0,
+    body: FilamentParameters,
-        name=None,
+) -> Filament:
-        vendor=Vendor(
+    db_item = await create_filament(
-            id=0,
+        db=db,
-            name="asdf",
+        density=body.density,
-            comment=None,
+        diameter=body.diameter,
-        ),
+        name=body.name,
-        material=None,
+        vendor_id=body.vendor_id,
-        price=None,
+        material=body.material,
-        density=1,
+        price=body.price,
-        diameter=1,
+        weight=body.weight,
-        weight=None,
+        spool_weight=body.spool_weight,
-        spool_weight=None,
+        article_number=body.article_number,
-        article_number=None,
+        comment=body.comment,
-        comment=None,
     )
 
+    return Filament.from_db(db_item)
+
 
 @router.put("/{filament_id}")
 async def update() -> Filament:
@@ -82,5 +107,5 @@ async def update() -> Filament:
 
 
 @router.delete("/{filament_id}")
-async def delete():
+async def delete() -> None:
     pass

spoolson/api/v1/models.py

@@ -5,12 +5,23 @@ from typing import Optional
 
 from pydantic import BaseModel, Field
 
+from spoolson.database import models
+
 
 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.")
@@ -31,6 +42,23 @@ class Filament(BaseModel):
     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.")
@@ -42,3 +70,18 @@ class Spool(BaseModel):
     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,
+        )

spoolson/api/v1/router.py

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

spoolson/database/database.py

@@ -1,7 +1,54 @@
-from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+"""SQLAlchemy database setup."""
 
-SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
+from collections.abc import AsyncGenerator
-# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
+from typing import Optional
 
-engine = create_async_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
-SessionLocal = async_sessionmaker(engine, autocommit=False, autoflush=False)
+
+from spoolson.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()

spoolson/database/filament.py

@@ -0,0 +1,43 @@
+"""Helper functions for interacting with filament database objects."""
+
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolson.database import models, vendor
+
+
+async def create_filament(
+    *,
+    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

spoolson/database/models.py

@@ -3,8 +3,9 @@
 from datetime import datetime
 from typing import Optional
 
-from sqlalchemy import Integer, String
+from sqlalchemy import ForeignKey, Integer, String
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+from sqlalchemy.sql import func
 
 
 class Base(DeclarativeBase):
@@ -17,14 +18,17 @@ class Vendor(Base):
     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__ = "material"
+    __tablename__ = "filament"
 
     id: Mapped[int] = mapped_column(primary_key=True, index=True)
     name: Mapped[Optional[str]] = mapped_column(String(64))
-    vendor: Mapped[Optional["Vendor"]] = relationship()
+    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()
@@ -41,10 +45,11 @@ class Spool(Base):
     __tablename__ = "spool"
 
     id: Mapped[int] = mapped_column(primary_key=True, index=True)
-    registered: Mapped[datetime] = mapped_column()
+    registered: Mapped[datetime] = mapped_column(default=func.now())
     first_used: Mapped[Optional[datetime]] = mapped_column()
     last_used: Mapped[Optional[datetime]] = mapped_column()
-    filament: Mapped["Filament"] = mapped_column(index=True)
+    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))

spoolson/database/vendor.py

@@ -0,0 +1,14 @@
+"""Helper functions for interacting with vendor database objects."""
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolson.database import models
+from spoolson.exceptions import ItemNotFoundError
+
+
+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

spoolson/exceptions.py

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

spoolson/main.py

@@ -1,13 +1,21 @@
 """Main entrypoint to the server."""
 
+
 import uvicorn
 from fastapi import FastAPI
 
-import spoolson.api.v1.router
+from spoolson.api.v1.router import app as v1_app
+from spoolson.database.database import setup_db
 
-app = FastAPI()
+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")
 
-app.mount("/api/v1", spoolson.api.v1.router.app)
 
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=8000)