first backend draft

backend/config.py

+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+    # we may want use DSN validators from pydantic.
+    # but there are no sqlite validators?
+    database_url: str = "sqlite:///./paketshop.db"
+
+    class Config:
+        env_file = ".env"
+
+
+settings = Settings()

backend/database.py

+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from .config import settings
+
+engine = create_engine(settings.database_url, connect_args={"check_same_thread": False})
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+
+def create_database():
+    Base.metadata.create_all(bind=engine)

backend/main.py

+from uuid import UUID
+
+from fastapi import Depends, FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from sqlalchemy.orm import Session
+
+from . import schemas, utils
+from .database import SessionLocal, create_database
+
+create_database()
+
+app = FastAPI()
+
+origins = ["http://127.0.0.1:8000", "http://localhost:3000", "http://localhost:3002"]
+
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# Dependency
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+@app.post("/item/prepare", response_model=schemas.Item)
+def add_item(item: schemas.ItemCreatePrepareShipping, db: Session = Depends(get_db)):
+    return utils.prepare_item_shipping(db, item)
+
+
+@app.get("/item/{item_uuid}", response_model=schemas.Item)
+def get_item(item_uuid: str, db: Session = Depends(get_db)):
+    item = utils.get_item_by_uuid(db, UUID(item_uuid))
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return utils.get_item_by_uuid(db, UUID(item_uuid))
+
+
+@app.get("/tag/{tag}", response_model=schemas.Item)
+def get_item_by_tag(tag: str, db: Session = Depends(get_db)):
+    item = utils.get_item_by_tag(db, tag)
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return utils.get_item_by_tag(db, tag)
+
+
+@app.get("/storages", response_model=list[schemas.Storage])
+def list_storages(db: Session = Depends(get_db)):
+    return utils.get_storages(db)
+
+
+@app.post("/checkin/{item_uuid}/at/{storage_name}", response_model=schemas.Item)
+def checkin_item_by_uuid(
+    item_uuid: str, storage_name: str, db: Session = Depends(get_db)
+):
+    item = utils.get_item_by_uuid(db, UUID(item_uuid))
+    if item is None:
+        raise HTTPException(status_code=404, detail="Item not found")
+    storage = utils.get_storage(db, storage_name)
+    if storage is None:
+        raise HTTPException(status_code=404, detail="Storage not found")
+    return utils.receive_item(db, item, storage)

backend/models.py

+from uuid import uuid4
+
+from sqlalchemy import Column as sql_Column
+from sqlalchemy import DateTime as sql_DateTime
+from sqlalchemy import ForeignKey as sql_ForeignKey
+from sqlalchemy import LargeBinary as sql_LargeBinary
+from sqlalchemy import String as sql_String
+from sqlalchemy import Uuid as sql_Uuid
+from sqlalchemy.orm import relationship as sql_relationship
+from sqlalchemy.sql.functions import now as sql_now
+
+from .database import Base as db_Base
+
+
+class Item(db_Base):
+    __tablename__ = "items"
+
+    uuid = sql_Column(sql_Uuid, primary_key=True, default=uuid4)
+    created_at = sql_Column(sql_DateTime(timezone=True), server_default=sql_now())
+
+    received_at = sql_Column(sql_DateTime(timezone=True), nullable=True, default=None)
+
+    addressee = sql_Column(sql_String(64), nullable=True, default=None)
+    team = sql_Column(sql_String(16), nullable=True, default=None)
+    images = sql_relationship("Image", back_populates="item")
+
+    deployed = sql_Column(sql_String(64), nullable=True, default=None)
+    deployed_at = sql_Column(sql_DateTime(timezone=True), nullable=True, default=None)
+
+    verification = sql_Column(sql_LargeBinary(57), nullable=True, default=None)
+    tag = sql_Column(sql_String(6), nullable=True, default=None)
+
+    storage = sql_Column(
+        sql_Uuid, sql_ForeignKey("store.name"), nullable=True, default=None
+    )
+    stored_at = sql_relationship("Storage", back_populates="items")
+
+
+class Storage(db_Base):
+    __tablename__ = "store"
+
+    name = sql_Column(sql_String(16), primary_key=True)
+    items = sql_relationship("Item", back_populates="stored_at")
+
+
+class Image(db_Base):
+    __tablename__ = "images"
+
+    uuid = sql_Column(sql_Uuid, primary_key=True, default=uuid4)
+    item_uuid = sql_Column(sql_Uuid, sql_ForeignKey("items.uuid"))
+    item = sql_relationship("Item", back_populates="images")

backend/schemas.py

+from datetime import datetime
+from typing import List, Union
+
+from pydantic import UUID4, BaseModel
+
+
+class Image(BaseModel):
+    uuid: UUID4
+
+    class Config:
+        orm_mode = True
+
+
+class Storage(BaseModel):
+    name: str
+    items: "List[Item]"
+
+    class Config:
+        orm_mode = True
+
+
+class ItemCreatePrepareShipping(BaseModel):
+    verification: bytes
+    addressee: Union[str, None] = None
+    team: Union[str, None] = None
+
+
+class ItemCreateByImageAtStorage(BaseModel):
+    image_uuid: UUID4
+    storage_uuid: UUID4
+
+
+class Item(BaseModel):
+    uuid: UUID4
+    created_at: datetime
+    received_at: Union[datetime, None] = None
+
+    addressee: Union[str, None] = None
+    team: Union[str, None] = None
+    images: List[Image] = []
+
+    deployed: Union[str, None] = None
+    deployed_at: Union[datetime, None] = None
+
+    tag: Union[str, None] = None
+
+    stored_at: Union[Storage, None] = None
+
+    class Config:
+        orm_mode = True
+
+
+Storage.update_forward_refs()

backend/utils.py

+from datetime import datetime
+from secrets import token_hex
+
+from pydantic import UUID4
+from sqlalchemy.orm import Session
+
+from . import models, schemas
+
+
+def get_item_by_uuid(db: Session, item_uuid: UUID4):
+    return db.get(models.Item, item_uuid)
+
+
+def get_item_by_tag(db: Session, item_tag: str):
+    return db.query(models.Item).filter(models.Item.tag == item_tag).first()
+
+
+def get_items_for_storage(db: Session, storage_name: str):
+    return db.query(models.Storage).get(models.Storage.name == storage_name).items
+
+
+def get_storage(db: Session, storage_name: str):
+    return db.get(models.Storage, storage_name)
+
+
+def get_storages(db: Session):
+    return db.query(models.Storage).all()
+
+
+def prepare_item_shipping(db: Session, item: schemas.ItemCreatePrepareShipping):
+    # we want the tag to be unique.
+    # FIXME: this may never finish.
+    tag = token_hex(3)
+    while db.query(models.Item).filter(models.Item.tag == tag).count() > 0:
+        tag = token_hex(3)
+    new_item = models.Item(**item.dict(), tag=tag)
+    db.add(new_item)
+    db.commit()
+    db.refresh(new_item)
+    return new_item
+
+
+def receive_item_with_image(db: Session, item: schemas.ItemCreateByImageAtStorage):
+    new_item = models.Item(storage_uuid=item.storage_uuid)
+    db.add(new_item)
+    db.commit()
+    db.refresh(new_item)
+    new_image = models.Image(item_uuid=new_item.uuid)
+    db.add(new_image)
+    db.commit()
+    return new_item
+
+
+def receive_item(db: Session, item: schemas.Item, storage: schemas.Storage):
+    item.received_at = datetime.now()
+    item.storage = storage.name
+    db.commit()
+    db.refresh(item)
+    return item