439330125e
- Fix unused imports in API endpoints - Add proper __all__ exports in model and schema modules - Add proper TYPE_CHECKING imports in models to prevent circular imports - Fix import order in migrations - Fix long lines in migration scripts - All ruff checks passing
118 lines
3.2 KiB
Python
118 lines
3.2 KiB
Python
import uuid
|
|
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.product import Product
|
|
from app.schemas.product import ProductCreate, ProductUpdate
|
|
|
|
|
|
class CRUDProduct(CRUDBase[Product, ProductCreate, ProductUpdate]):
|
|
"""
|
|
CRUD operations for Product model.
|
|
"""
|
|
def get_by_sku(self, db: Session, *, sku: str) -> Optional[Product]:
|
|
"""
|
|
Get a product by SKU.
|
|
"""
|
|
return db.query(Product).filter(Product.sku == sku).first()
|
|
|
|
def create_with_owner(
|
|
self, db: Session, *, obj_in: ProductCreate, owner_id: str
|
|
) -> Product:
|
|
"""
|
|
Create a new product with owner ID.
|
|
"""
|
|
db_obj = Product(
|
|
id=str(uuid.uuid4()),
|
|
name=obj_in.name,
|
|
description=obj_in.description,
|
|
sku=obj_in.sku,
|
|
price=obj_in.price,
|
|
cost=obj_in.cost,
|
|
quantity=obj_in.quantity,
|
|
reorder_level=obj_in.reorder_level,
|
|
category_id=obj_in.category_id,
|
|
supplier_id=obj_in.supplier_id,
|
|
owner_id=owner_id,
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_multi_by_owner(
|
|
self, db: Session, *, owner_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[Product]:
|
|
"""
|
|
Get multiple products by owner ID.
|
|
"""
|
|
return (
|
|
db.query(Product)
|
|
.filter(Product.owner_id == owner_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_multi_by_category(
|
|
self, db: Session, *, category_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[Product]:
|
|
"""
|
|
Get multiple products by category ID.
|
|
"""
|
|
return (
|
|
db.query(Product)
|
|
.filter(Product.category_id == category_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_multi_by_supplier(
|
|
self, db: Session, *, supplier_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[Product]:
|
|
"""
|
|
Get multiple products by supplier ID.
|
|
"""
|
|
return (
|
|
db.query(Product)
|
|
.filter(Product.supplier_id == supplier_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_low_stock_products(
|
|
self, db: Session, *, owner_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[Product]:
|
|
"""
|
|
Get products with stock levels below reorder level.
|
|
"""
|
|
return (
|
|
db.query(Product)
|
|
.filter(Product.owner_id == owner_id)
|
|
.filter(Product.quantity <= Product.reorder_level)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_out_of_stock_products(
|
|
self, db: Session, *, owner_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[Product]:
|
|
"""
|
|
Get products with zero stock.
|
|
"""
|
|
return (
|
|
db.query(Product)
|
|
.filter(Product.owner_id == owner_id)
|
|
.filter(Product.quantity == 0)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
product = CRUDProduct(Product) |