4dc182713c
- Created project structure with FastAPI framework - Set up SQLite database with SQLAlchemy ORM - Implemented models: User, Item, Supplier, Transaction - Created CRUD operations for all models - Added API endpoints for all resources - Implemented JWT authentication and authorization - Added Alembic migration scripts - Created health endpoint - Updated documentation generated with BackendIM... (backend.im)
49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
from typing import List, Optional
|
|
|
|
from fastapi.encoders import jsonable_encoder
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.item import Item
|
|
from app.schemas.item import ItemCreate, ItemUpdate
|
|
|
|
|
|
class CRUDItem(CRUDBase[Item, ItemCreate, ItemUpdate]):
|
|
def get_by_sku(self, db: Session, *, sku: str) -> Optional[Item]:
|
|
return db.query(Item).filter(Item.sku == sku).first()
|
|
|
|
def get_by_supplier(
|
|
self, db: Session, *, supplier_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Item]:
|
|
return (
|
|
db.query(Item)
|
|
.filter(Item.supplier_id == supplier_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_category(
|
|
self, db: Session, *, category: str, skip: int = 0, limit: int = 100
|
|
) -> List[Item]:
|
|
return (
|
|
db.query(Item)
|
|
.filter(Item.category == category)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def update_quantity(
|
|
self, db: Session, *, item_id: int, quantity_change: int
|
|
) -> Item:
|
|
item = self.get(db, id=item_id)
|
|
if item:
|
|
item.quantity += quantity_change
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
item = CRUDItem(Item) |