e8172f2bc2
- Set up project structure with FastAPI and SQLite - Created database models for users, categories, suppliers, items, and stock transactions - Implemented Alembic for database migrations with proper absolute paths - Built comprehensive CRUD operations for all entities - Added JWT-based authentication and authorization system - Created RESTful API endpoints for all inventory operations - Implemented search, filtering, and low stock alerts - Added health check endpoint and base URL response - Configured CORS for all origins - Set up Ruff for code linting and formatting - Updated README with comprehensive documentation and usage examples The system provides complete inventory management functionality for small businesses including product tracking, supplier management, stock transactions, and reporting. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from app.crud.base import CRUDBase
|
|
from app.models.item import Item
|
|
from app.schemas.item import ItemCreate, ItemUpdate
|
|
|
|
|
|
class CRUDItem(CRUDBase[Item, ItemCreate, ItemUpdate]):
|
|
def get_by_sku(self, db: Session, *, sku: str) -> Optional[Item]:
|
|
return db.query(Item).filter(Item.sku == sku).first()
|
|
|
|
def get_by_barcode(self, db: Session, *, barcode: str) -> Optional[Item]:
|
|
return db.query(Item).filter(Item.barcode == barcode).first()
|
|
|
|
def get_by_category(
|
|
self, db: Session, *, category_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Item]:
|
|
return (
|
|
db.query(Item)
|
|
.filter(Item.category_id == category_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_supplier(
|
|
self, db: Session, *, supplier_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Item]:
|
|
return (
|
|
db.query(Item)
|
|
.filter(Item.supplier_id == supplier_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_low_stock_items(
|
|
self, db: Session, skip: int = 0, limit: int = 100
|
|
) -> List[Item]:
|
|
return (
|
|
db.query(Item)
|
|
.filter(Item.quantity_in_stock <= Item.reorder_point)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_active_items(
|
|
self, db: Session, skip: int = 0, limit: int = 100
|
|
) -> List[Item]:
|
|
return db.query(Item).filter(Item.is_active).offset(skip).limit(limit).all()
|
|
|
|
def search_items(
|
|
self, db: Session, *, query: str, skip: int = 0, limit: int = 100
|
|
) -> List[Item]:
|
|
return (
|
|
db.query(Item)
|
|
.filter(
|
|
Item.name.contains(query)
|
|
| Item.description.contains(query)
|
|
| Item.sku.contains(query)
|
|
)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
item = CRUDItem(Item)
|