Automated Action b078a91dd3 Implement simple ecommerce API with FastAPI and SQLite
- Setup project structure and FastAPI application
- Create SQLAlchemy models for users, products, carts, and orders
- Implement Alembic migrations
- Add CRUD operations and endpoints for all resources
- Setup authentication with JWT
- Add role-based access control
- Update documentation
2025-06-12 17:15:18 +00:00

72 lines
1.9 KiB
Python

from typing import List, Optional
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.product import Product
from app.schemas.product import ProductCreate, ProductUpdate
class CRUDProduct(CRUDBase[Product, ProductCreate, ProductUpdate]):
def search_products(
self, db: Session, *, query: str, skip: int = 0, limit: int = 100
) -> List[Product]:
"""
Search for products by name or description.
Args:
db: Database session
query: Search query
skip: Number of records to skip
limit: Maximum number of records to return
Returns:
List of products matching the search query
"""
search_pattern = f"%{query}%"
return (
db.query(self.model)
.filter(
(self.model.name.ilike(search_pattern))
| (self.model.description.ilike(search_pattern))
)
.offset(skip)
.limit(limit)
.all()
)
def get_product_by_name(self, db: Session, *, name: str) -> Optional[Product]:
"""
Get a product by its exact name.
Args:
db: Database session
name: Product name
Returns:
The product or None if not found
"""
return db.query(self.model).filter(self.model.name == name).first()
def update_stock(self, db: Session, *, product_id: int, quantity: int) -> Product:
"""
Update product stock quantity.
Args:
db: Database session
product_id: Product ID
quantity: Quantity to add (positive) or subtract (negative)
Returns:
The updated product
"""
product = self.get(db, id=product_id)
if product:
product.stock += quantity
db.commit()
db.refresh(product)
return product
product = CRUDProduct(Product)