5935f302dc
- Set up project structure and FastAPI application - Create database models with SQLAlchemy - Implement authentication with JWT - Add CRUD operations for products, inventory, categories - Implement purchase order and sales functionality - Create reporting endpoints - Set up Alembic for database migrations - Add comprehensive documentation in README.md
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.inventory import Inventory, InventoryTransaction
|
|
from app.models.product import Product
|
|
from app.schemas.inventory import InventoryCreate, InventoryUpdate, InventoryAdjustment
|
|
|
|
|
|
class CRUDInventory(CRUDBase[Inventory, InventoryCreate, InventoryUpdate]):
|
|
def get_by_product_and_location(
|
|
self, db: Session, *, product_id: int, location: Optional[str] = None
|
|
) -> Optional[Inventory]:
|
|
"""Get inventory by product and location"""
|
|
query = db.query(Inventory).filter(Inventory.product_id == product_id)
|
|
if location:
|
|
query = query.filter(Inventory.location == location)
|
|
return query.first()
|
|
|
|
def get_product_inventory(
|
|
self, db: Session, *, product_id: int
|
|
) -> List[Inventory]:
|
|
"""Get all inventory records for a product"""
|
|
return db.query(Inventory).filter(Inventory.product_id == product_id).all()
|
|
|
|
def get_total_product_quantity(
|
|
self, db: Session, *, product_id: int
|
|
) -> int:
|
|
"""Get total inventory quantity for a product across all locations"""
|
|
result = db.query(
|
|
func.sum(Inventory.quantity).label("total")
|
|
).filter(
|
|
Inventory.product_id == product_id
|
|
).scalar()
|
|
return result or 0
|
|
|
|
def adjust_inventory(
|
|
self, db: Session, *, adjustment: InventoryAdjustment
|
|
) -> Dict[str, Any]:
|
|
"""Adjust inventory quantity and record the transaction"""
|
|
# Find or create inventory record
|
|
inventory = self.get_by_product_and_location(
|
|
db, product_id=adjustment.product_id, location=adjustment.location
|
|
)
|
|
|
|
if not inventory:
|
|
# Create new inventory record if it doesn't exist
|
|
inventory = Inventory(
|
|
product_id=adjustment.product_id,
|
|
quantity=0, # Start with 0, will add adjustment below
|
|
location=adjustment.location
|
|
)
|
|
db.add(inventory)
|
|
db.flush()
|
|
|
|
# Update quantity
|
|
inventory.quantity += adjustment.quantity
|
|
|
|
# Ensure quantity is never negative
|
|
if inventory.quantity < 0:
|
|
inventory.quantity = 0
|
|
|
|
# Record transaction
|
|
transaction = InventoryTransaction(
|
|
product_id=adjustment.product_id,
|
|
quantity=adjustment.quantity,
|
|
transaction_type="adjustment",
|
|
reason=adjustment.reason,
|
|
location=adjustment.location
|
|
)
|
|
db.add(transaction)
|
|
|
|
db.commit()
|
|
db.refresh(inventory)
|
|
db.refresh(transaction)
|
|
|
|
return {
|
|
"inventory": inventory,
|
|
"transaction": transaction
|
|
}
|
|
|
|
|
|
inventory = CRUDInventory(Inventory) |