0e00013b7f
- Created FastAPI application with product catalog functionality - Implemented CRUD operations for products - Added SQLAlchemy models and Pydantic schemas - Set up SQLite database with Alembic migrations - Added health check endpoint - Updated README with project documentation generated with BackendIM... (backend.im)
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.database import get_db
|
|
from app.models.product import Product as ProductModel
|
|
from app.schemas.product import Product, ProductCreate, ProductUpdate
|
|
from app.crud import product as product_crud
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[Product])
|
|
def get_products(
|
|
category: Optional[str] = None,
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=100),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Retrieve all products with optional filtering by category.
|
|
"""
|
|
products = product_crud.get_products(db, skip=skip, limit=limit, category=category)
|
|
return products
|
|
|
|
|
|
@router.post("/", response_model=Product, status_code=status.HTTP_201_CREATED)
|
|
def create_product(
|
|
product: ProductCreate,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Create a new product.
|
|
"""
|
|
db_product = product_crud.get_product_by_sku(db, sku=product.sku)
|
|
if db_product:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Product with SKU {product.sku} already exists",
|
|
)
|
|
return product_crud.create_product(db=db, product=product)
|
|
|
|
|
|
@router.get("/{product_id}", response_model=Product)
|
|
def get_product(
|
|
product_id: int,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Get a specific product by ID.
|
|
"""
|
|
db_product = product_crud.get_product(db, product_id=product_id)
|
|
if db_product is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Product with ID {product_id} not found",
|
|
)
|
|
return db_product
|
|
|
|
|
|
@router.put("/{product_id}", response_model=Product)
|
|
def update_product(
|
|
product_id: int,
|
|
product_update: ProductUpdate,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Update a product.
|
|
"""
|
|
db_product = product_crud.get_product(db, product_id=product_id)
|
|
if db_product is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Product with ID {product_id} not found",
|
|
)
|
|
|
|
# If SKU is being updated, check if it already exists
|
|
if product_update.sku is not None and product_update.sku != db_product.sku:
|
|
existing_product = product_crud.get_product_by_sku(db, sku=product_update.sku)
|
|
if existing_product:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Product with SKU {product_update.sku} already exists",
|
|
)
|
|
|
|
return product_crud.update_product(db=db, db_product=db_product, product_update=product_update)
|
|
|
|
|
|
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_product(
|
|
product_id: int,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Delete a product.
|
|
"""
|
|
db_product = product_crud.get_product(db, product_id=product_id)
|
|
if db_product is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Product with ID {product_id} not found",
|
|
)
|
|
|
|
product_crud.delete_product(db=db, db_product=db_product)
|
|
return None |