d340d37ac9
- Set up project structure with FastAPI and SQLite - Create database models for inventory management - Set up SQLAlchemy and Alembic for database migrations - Create initial database migrations - Implement CRUD operations for products, categories, suppliers - Implement stock movement tracking and inventory management - Add authentication and user management - Add API endpoints for all entities - Add health check endpoint - Update README with project information and usage instructions
54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.stock_movement import StockMovement, MovementType
|
|
from app.schemas.stock_movement import StockMovementCreate
|
|
from app.crud.crud_product import product as product_crud
|
|
|
|
|
|
class CRUDStockMovement(CRUDBase[StockMovement, StockMovementCreate, StockMovementCreate]):
|
|
def create_with_product_update(
|
|
self, db: Session, *, obj_in: StockMovementCreate, created_by: Optional[str] = None
|
|
) -> StockMovement:
|
|
# Create the stock movement
|
|
obj_in_data = obj_in.dict()
|
|
obj_in_data["created_by"] = created_by
|
|
db_obj = super().create(db, obj_in=obj_in)
|
|
|
|
# Update product stock level
|
|
quantity = obj_in.quantity
|
|
if obj_in.movement_type in [MovementType.SALE, MovementType.ADJUSTMENT]:
|
|
if obj_in.quantity > 0:
|
|
quantity = -quantity # Negate for outgoing stock
|
|
|
|
product_crud.update_stock(db, product_id=obj_in.product_id, quantity=quantity)
|
|
|
|
return db_obj
|
|
|
|
def get_by_product(
|
|
self, db: Session, *, product_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[StockMovement]:
|
|
return (
|
|
db.query(StockMovement)
|
|
.filter(StockMovement.product_id == product_id)
|
|
.order_by(StockMovement.created_at.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_movement_type(
|
|
self, db: Session, *, movement_type: MovementType, skip: int = 0, limit: int = 100
|
|
) -> List[StockMovement]:
|
|
return (
|
|
db.query(StockMovement)
|
|
.filter(StockMovement.movement_type == movement_type)
|
|
.order_by(StockMovement.created_at.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
stock_movement = CRUDStockMovement(StockMovement) |