80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.models.business import Product, ProductCategory
|
|
|
|
|
|
class ProductRepository:
|
|
def list_categories(self, session: Session, filters: dict) -> list[ProductCategory]:
|
|
stmt = select(ProductCategory).where(ProductCategory.deleted == 0)
|
|
|
|
if filters.get("category_name"):
|
|
stmt = stmt.where(ProductCategory.category_name.contains(filters["category_name"]))
|
|
if filters.get("status") is not None:
|
|
stmt = stmt.where(ProductCategory.status == filters["status"])
|
|
|
|
stmt = stmt.order_by(ProductCategory.sort_no.asc(), ProductCategory.id.asc())
|
|
return list(session.execute(stmt).scalars())
|
|
|
|
def get_category(self, session: Session, category_id: int) -> ProductCategory | None:
|
|
stmt = select(ProductCategory).where(ProductCategory.id == category_id, ProductCategory.deleted == 0)
|
|
return session.execute(stmt).scalar_one_or_none()
|
|
|
|
def get_category_by_code(self, session: Session, category_code: str) -> ProductCategory | None:
|
|
stmt = select(ProductCategory).where(
|
|
ProductCategory.category_code == category_code,
|
|
ProductCategory.deleted == 0,
|
|
)
|
|
return session.execute(stmt).scalar_one_or_none()
|
|
|
|
def create_category(self, session: Session, payload: dict) -> ProductCategory:
|
|
category = ProductCategory(**payload)
|
|
session.add(category)
|
|
session.flush()
|
|
return category
|
|
|
|
def update_category(self, session: Session, category: ProductCategory, payload: dict) -> ProductCategory:
|
|
for key, value in payload.items():
|
|
setattr(category, key, value)
|
|
session.add(category)
|
|
session.flush()
|
|
return category
|
|
|
|
def list_products(self, session: Session, filters: dict) -> list[Product]:
|
|
stmt = select(Product).where(Product.deleted == 0)
|
|
|
|
if filters.get("product_name"):
|
|
stmt = stmt.where(Product.product_name.contains(filters["product_name"]))
|
|
if filters.get("specification"):
|
|
stmt = stmt.where(Product.specification.contains(filters["specification"]))
|
|
if filters.get("category_id") is not None:
|
|
stmt = stmt.where(Product.category_id == filters["category_id"])
|
|
if filters.get("status") is not None:
|
|
stmt = stmt.where(Product.status == filters["status"])
|
|
|
|
stmt = stmt.order_by(Product.id.desc())
|
|
return list(session.execute(stmt).scalars())
|
|
|
|
def get_product(self, session: Session, product_id: int) -> Product | None:
|
|
stmt = select(Product).where(Product.id == product_id, Product.deleted == 0)
|
|
return session.execute(stmt).scalar_one_or_none()
|
|
|
|
def get_product_by_name_and_specification(
|
|
self,
|
|
session: Session,
|
|
product_name: str,
|
|
specification: str,
|
|
) -> Product | None:
|
|
stmt = select(Product).where(
|
|
Product.product_name == product_name,
|
|
Product.specification == specification,
|
|
Product.deleted == 0,
|
|
)
|
|
return session.execute(stmt).scalar_one_or_none()
|
|
|
|
def create_product(self, session: Session, payload: dict) -> Product:
|
|
product = Product(**payload)
|
|
session.add(product)
|
|
session.flush()
|
|
return product
|