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
69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.product import Product
|
|
from app.models.inventory import Inventory
|
|
from app.schemas.product import ProductCreate, ProductUpdate
|
|
|
|
|
|
class CRUDProduct(CRUDBase[Product, ProductCreate, ProductUpdate]):
|
|
def get_with_inventory(self, db: Session, id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get product with inventory quantities"""
|
|
product = db.query(Product).filter(Product.id == id).first()
|
|
|
|
if not product:
|
|
return None
|
|
|
|
# Get total inventory for this product
|
|
inventory = db.query(
|
|
func.sum(Inventory.quantity).label("total_quantity")
|
|
).filter(
|
|
Inventory.product_id == id
|
|
).scalar() or 0
|
|
|
|
result = {
|
|
**product.__dict__,
|
|
"total_quantity": inventory,
|
|
"available_quantity": inventory # For now they're the same since we don't track reserved items
|
|
}
|
|
|
|
return result
|
|
|
|
def get_multi_with_inventory(
|
|
self, db: Session, *, skip: int = 0, limit: int = 100
|
|
) -> List[Dict[str, Any]]:
|
|
"""Get multiple products with their inventory quantities"""
|
|
products = db.query(Product).offset(skip).limit(limit).all()
|
|
|
|
# Get inventory counts for all products in one query
|
|
inventory_counts = dict(
|
|
db.query(
|
|
Inventory.product_id,
|
|
func.sum(Inventory.quantity).label("total")
|
|
).group_by(
|
|
Inventory.product_id
|
|
).all()
|
|
)
|
|
|
|
result = []
|
|
for product in products:
|
|
total_quantity = inventory_counts.get(product.id, 0)
|
|
|
|
result.append({
|
|
**product.__dict__,
|
|
"total_quantity": total_quantity,
|
|
"available_quantity": total_quantity # For now they're the same
|
|
})
|
|
|
|
return result
|
|
|
|
def get_by_sku(self, db: Session, *, sku: str) -> Optional[Product]:
|
|
return db.query(Product).filter(Product.sku == sku).first()
|
|
|
|
def get_by_barcode(self, db: Session, *, barcode: str) -> Optional[Product]:
|
|
return db.query(Product).filter(Product.barcode == barcode).first()
|
|
|
|
|
|
product = CRUDProduct(Product) |