109 lines
2.8 KiB
Python
109 lines
2.8 KiB
Python
from typing import Any, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import crud, models, schemas
|
|
from app.api import deps
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[schemas.Product])
|
|
def read_products(
|
|
db: Session = Depends(deps.get_db),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
name: Optional[str] = None,
|
|
category_id: Optional[int] = None,
|
|
min_price: Optional[float] = None,
|
|
max_price: Optional[float] = None,
|
|
in_stock: Optional[bool] = None,
|
|
) -> Any:
|
|
"""
|
|
Retrieve products with filtering options.
|
|
"""
|
|
filter_params = schemas.ProductFilterParams(
|
|
name=name,
|
|
category_id=category_id,
|
|
min_price=min_price,
|
|
max_price=max_price,
|
|
in_stock=in_stock,
|
|
)
|
|
products = crud.product.search_products(
|
|
db, filter_params=filter_params, skip=skip, limit=limit
|
|
)
|
|
return products
|
|
|
|
|
|
@router.post("/", response_model=schemas.Product)
|
|
def create_product(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
product_in: schemas.ProductCreate,
|
|
current_user: models.User = Depends(deps.get_current_admin_user),
|
|
) -> Any:
|
|
"""
|
|
Create new product (admin only).
|
|
"""
|
|
product = crud.product.create(db, obj_in=product_in)
|
|
return product
|
|
|
|
|
|
@router.get("/{id}", response_model=schemas.Product)
|
|
def read_product(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
id: int,
|
|
) -> Any:
|
|
"""
|
|
Get product by ID.
|
|
"""
|
|
product = crud.product.get(db, id=id)
|
|
if not product:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Product not found",
|
|
)
|
|
return product
|
|
|
|
|
|
@router.put("/{id}", response_model=schemas.Product)
|
|
def update_product(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
id: int,
|
|
product_in: schemas.ProductUpdate,
|
|
current_user: models.User = Depends(deps.get_current_admin_user),
|
|
) -> Any:
|
|
"""
|
|
Update a product (admin only).
|
|
"""
|
|
product = crud.product.get(db, id=id)
|
|
if not product:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Product not found",
|
|
)
|
|
product = crud.product.update(db, db_obj=product, obj_in=product_in)
|
|
return product
|
|
|
|
|
|
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
|
|
def delete_product(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
id: int,
|
|
current_user: models.User = Depends(deps.get_current_admin_user),
|
|
) -> None:
|
|
"""
|
|
Delete a product (admin only).
|
|
"""
|
|
product = crud.product.get(db, id=id)
|
|
if not product:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Product not found",
|
|
)
|
|
crud.product.remove(db, id=id)
|