447799c9a0
Features implemented: - Product management with CRUD operations - Category and supplier management - Stock movement tracking with automatic updates - Inventory reports and analytics - SQLite database with Alembic migrations - Health monitoring endpoints - CORS configuration for API access - Comprehensive API documentation - Code quality with Ruff linting and formatting The system provides a complete backend solution for small business inventory management with proper database relationships, stock tracking, and reporting capabilities.
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from app.db.session import get_db
|
|
from app.models.stock_movement import StockMovement
|
|
from app.schemas.stock_movement import (
|
|
StockMovement as StockMovementSchema,
|
|
StockMovementCreate,
|
|
)
|
|
|
|
router = APIRouter(prefix="/stock-movements", tags=["stock-movements"])
|
|
|
|
|
|
@router.get("/", response_model=List[StockMovementSchema])
|
|
def get_stock_movements(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
product_id: Optional[int] = None,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
query = db.query(StockMovement)
|
|
|
|
if product_id:
|
|
query = query.filter(StockMovement.product_id == product_id)
|
|
|
|
return (
|
|
query.order_by(StockMovement.created_at.desc()).offset(skip).limit(limit).all()
|
|
)
|
|
|
|
|
|
@router.get("/{movement_id}", response_model=StockMovementSchema)
|
|
def get_stock_movement(movement_id: int, db: Session = Depends(get_db)):
|
|
movement = db.query(StockMovement).filter(StockMovement.id == movement_id).first()
|
|
if not movement:
|
|
raise HTTPException(status_code=404, detail="Stock movement not found")
|
|
return movement
|
|
|
|
|
|
@router.post("/", response_model=StockMovementSchema)
|
|
def create_stock_movement(movement: StockMovementCreate, db: Session = Depends(get_db)):
|
|
db_movement = StockMovement(**movement.dict())
|
|
db.add(db_movement)
|
|
db.commit()
|
|
db.refresh(db_movement)
|
|
return db_movement
|