100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.api.deps import get_product_service
|
|
from backend.app.db import get_db_session
|
|
from backend.app.schemas.common import success_payload
|
|
from backend.app.schemas.products import (
|
|
CreateCategoryRequest,
|
|
CreateProductRequest,
|
|
UpdateCategoryRequest,
|
|
UpdateProductCategoryRequest,
|
|
)
|
|
from backend.app.services.product_service import ProductService
|
|
|
|
router = APIRouter(tags=["products"])
|
|
|
|
|
|
@router.get("/api/product-categories")
|
|
def list_categories(
|
|
category_name: str | None = Query(default=None),
|
|
status: int | None = Query(default=None),
|
|
page_no: int = Query(default=1),
|
|
page_size: int = Query(default=20),
|
|
product_service: ProductService = Depends(get_product_service),
|
|
session: Session = Depends(get_db_session),
|
|
) -> dict:
|
|
result = product_service.list_categories(
|
|
session=session,
|
|
filters={"category_name": category_name, "status": status},
|
|
)
|
|
start = max(page_no - 1, 0) * page_size
|
|
result["list"] = result["list"][start : start + page_size]
|
|
result["page_no"] = page_no
|
|
result["page_size"] = page_size
|
|
return success_payload(result)
|
|
|
|
|
|
@router.post("/api/product-categories")
|
|
def create_category(
|
|
payload: CreateCategoryRequest,
|
|
product_service: ProductService = Depends(get_product_service),
|
|
session: Session = Depends(get_db_session),
|
|
) -> dict:
|
|
return success_payload(product_service.create_category(payload.model_dump(), session))
|
|
|
|
|
|
@router.put("/api/product-categories/{category_id}")
|
|
def update_category(
|
|
category_id: int,
|
|
payload: UpdateProductCategoryRequest,
|
|
product_service: ProductService = Depends(get_product_service),
|
|
session: Session = Depends(get_db_session),
|
|
) -> dict:
|
|
return success_payload(product_service.update_category(category_id, payload.model_dump(), session))
|
|
|
|
|
|
@router.get("/api/products")
|
|
def list_products(
|
|
product_name: str | None = Query(default=None),
|
|
specification: str | None = Query(default=None),
|
|
category_id: int | None = Query(default=None),
|
|
status: int | None = Query(default=None),
|
|
page_no: int = Query(default=1),
|
|
page_size: int = Query(default=20),
|
|
product_service: ProductService = Depends(get_product_service),
|
|
session: Session = Depends(get_db_session),
|
|
) -> dict:
|
|
result = product_service.list_products(
|
|
{
|
|
"product_name": product_name,
|
|
"specification": specification,
|
|
"category_id": category_id,
|
|
"status": status,
|
|
},
|
|
session,
|
|
)
|
|
start = max(page_no - 1, 0) * page_size
|
|
result["list"] = result["list"][start : start + page_size]
|
|
result["page_no"] = page_no
|
|
result["page_size"] = page_size
|
|
return success_payload(result)
|
|
|
|
|
|
@router.post("/api/products")
|
|
def create_product(
|
|
payload: CreateProductRequest,
|
|
product_service: ProductService = Depends(get_product_service),
|
|
session: Session = Depends(get_db_session),
|
|
) -> dict:
|
|
return success_payload(product_service.create_product(payload.model_dump(), session))
|
|
|
|
|
|
@router.get("/api/products/{product_id}")
|
|
def get_product(
|
|
product_id: int,
|
|
product_service: ProductService = Depends(get_product_service),
|
|
session: Session = Depends(get_db_session),
|
|
) -> dict:
|
|
return success_payload(product_service.get_product(product_id, session))
|