feat: core inventory API and secure operation routers

This commit is contained in:
Daniel Bedeleanu
2026-04-10 13:47:10 +03:00
parent 46bd6ae733
commit e4b0afb50d
6 changed files with 183 additions and 0 deletions

View File

@@ -1,12 +1,26 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from . import models
from .database import engine
from .routers import items, operations
# Create the database tables
models.Base.metadata.create_all(bind=engine)
app = FastAPI(title="Inventory PWA API", version="0.1.0")
# Setup Cross-Origin for Client interaction (PWA)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(items.router)
app.include_router(operations.router)
@app.get("/")
def read_root():
return {"message": "Inventory API is running"}

View File

@@ -0,0 +1 @@
# Empty init file

46
backend/routers/items.py Normal file
View File

@@ -0,0 +1,46 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from .. import models, schemas
from ..database import get_db
router = APIRouter(
prefix="/items",
tags=["Items"]
)
@router.get("/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = db.query(models.Item).offset(skip).limit(limit).all()
return items
@router.get("/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(models.Item).filter(models.Item.id == item_id).first()
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
@router.post("/", response_model=schemas.Item, status_code=status.HTTP_201_CREATED)
def create_item(item: schemas.ItemCreate, user_id: int, db: Session = Depends(get_db)):
# Check if barcode exists
db_item = db.query(models.Item).filter(models.Item.barcode == item.barcode).first()
if db_item:
raise HTTPException(status_code=400, detail="Barcode already registered")
db_item = models.Item(**item.model_dump())
db.add(db_item)
db.commit()
db.refresh(db_item)
# Audit log the creation
audit = models.AuditLog(
user_id=user_id,
action="CREATE_ITEM",
target_item_id=db_item.id,
quantity_change=item.quantity
)
db.add(audit)
db.commit()
return db_item

View File

@@ -0,0 +1,62 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .. import models, schemas
from ..database import get_db
router = APIRouter(
prefix="/operations",
tags=["Operations"]
)
@router.post("/check-in", response_model=schemas.Item)
def check_in_item(op: schemas.OperationCreate, db: Session = Depends(get_db)):
if op.quantity <= 0:
raise HTTPException(status_code=400, detail="Quantity must be greater than zero")
item = db.query(models.Item).filter(models.Item.barcode == op.barcode).first()
if not item:
raise HTTPException(status_code=404, detail="Item not found. Register item first.")
# Update quantity
item.quantity += op.quantity
# Create Mandatory Audit Log
audit = models.AuditLog(
user_id=op.user_id,
action="CHECK_IN",
target_item_id=item.id,
quantity_change=op.quantity
)
db.add(audit)
db.commit()
db.refresh(item)
return item
@router.post("/check-out", response_model=schemas.Item)
def check_out_item(op: schemas.OperationCreate, db: Session = Depends(get_db)):
if op.quantity <= 0:
raise HTTPException(status_code=400, detail="Quantity must be greater than zero")
item = db.query(models.Item).filter(models.Item.barcode == op.barcode).first()
if not item:
raise HTTPException(status_code=404, detail="Item not found")
if item.quantity < op.quantity:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Insufficient stock")
# Update quantity
item.quantity -= op.quantity
# Create Mandatory Audit Log
audit = models.AuditLog(
user_id=op.user_id,
action="CHECK_OUT",
target_item_id=item.id,
quantity_change=-op.quantity
)
db.add(audit)
db.commit()
db.refresh(item)
return item

40
backend/schemas.py Normal file
View File

@@ -0,0 +1,40 @@
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime
# --- Items ---
class ItemBase(BaseModel):
name: str
category: str
barcode: str
quantity: float = 0.0
min_quantity: float = 1.0
image_url: Optional[str] = None
labels_data: Optional[str] = None
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int
class Config:
from_attributes = True
# --- Operations (Check-in/Check-out Validation) ---
class OperationCreate(BaseModel):
barcode: str
quantity: float
user_id: int
# --- Audit Logs ---
class AuditLogResponse(BaseModel):
id: int
timestamp: datetime
user_id: int
action: str
target_item_id: Optional[int]
quantity_change: Optional[float]
class Config:
from_attributes = True

View File

@@ -9,3 +9,23 @@ Each entry MUST be formatted chronologically.
- `path/to/file`
**Test Results:** (if applicable)
### [2026-04-10 13:44] Backend Foundation Setup
**Purpose:** Initialized the Phase 1 backend using Python and SQLite to prepare the environment for API routing.
**Modified Files:**
- `backend/requirements.txt` (NEW)
- `backend/database.py` (NEW)
- `backend/models.py` (NEW)
- `backend/main.py` (NEW)
- `.gitignore` (NEW)
**Test Results:** Installed properly using Homebrew Python venv on Mac. SQLAlchemy models parsed appropriately.
### [2026-04-10 13:47] Core Inventory API (Phase 2)
**Purpose:** Implemented the Pydantic schemas and standard CRUD endpoints for Inventory logic. Built safe Check-in/Check-out operations enforcing immutable Audit Log triggers.
**Modified Files:**
- `backend/schemas.py` (NEW)
- `backend/routers/__init__.py` (NEW)
- `backend/routers/items.py` (NEW)
- `backend/routers/operations.py` (NEW)
- `backend/main.py` (UPDATED)