a28e115e4f
- Set up project structure with FastAPI and SQLite - Create models for products and cart items - Implement schemas for API request/response validation - Add database migrations with Alembic - Create service layer for business logic - Implement API endpoints for cart operations - Add health endpoint for monitoring - Update documentation - Fix linting issues
87 lines
2.1 KiB
Python
87 lines
2.1 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.product import Product
|
|
from app.schemas.product import ProductCreate, ProductUpdate
|
|
|
|
|
|
def get_products(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
is_active: Optional[bool] = None
|
|
) -> List[Product]:
|
|
"""
|
|
Get a list of products with pagination and optional filtering.
|
|
"""
|
|
query = db.query(Product)
|
|
|
|
if is_active is not None:
|
|
query = query.filter(Product.is_active == is_active)
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
|
|
def get_product_by_id(db: Session, product_id: int) -> Optional[Product]:
|
|
"""
|
|
Get a product by its ID.
|
|
"""
|
|
return db.query(Product).filter(Product.id == product_id).first()
|
|
|
|
|
|
def create_product(db: Session, product_data: ProductCreate) -> Product:
|
|
"""
|
|
Create a new product.
|
|
"""
|
|
db_product = Product(**product_data.dict())
|
|
db.add(db_product)
|
|
db.commit()
|
|
db.refresh(db_product)
|
|
return db_product
|
|
|
|
|
|
def update_product(
|
|
db: Session,
|
|
product_id: int,
|
|
product_data: ProductUpdate
|
|
) -> Optional[Product]:
|
|
"""
|
|
Update an existing product.
|
|
"""
|
|
db_product = get_product_by_id(db, product_id)
|
|
if not db_product:
|
|
return None
|
|
|
|
# Update only provided fields
|
|
update_data = product_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_product, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_product)
|
|
return db_product
|
|
|
|
|
|
def delete_product(db: Session, product_id: int) -> bool:
|
|
"""
|
|
Delete a product by its ID.
|
|
"""
|
|
db_product = get_product_by_id(db, product_id)
|
|
if not db_product:
|
|
return False
|
|
|
|
db.delete(db_product)
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def get_product_count(db: Session, is_active: Optional[bool] = None) -> int:
|
|
"""
|
|
Get the total count of products with optional filtering.
|
|
"""
|
|
query = db.query(Product)
|
|
|
|
if is_active is not None:
|
|
query = query.filter(Product.is_active == is_active)
|
|
|
|
return query.count() |