5935f302dc
- Set up project structure and FastAPI application - Create database models with SQLAlchemy - Implement authentication with JWT - Add CRUD operations for products, inventory, categories - Implement purchase order and sales functionality - Create reporting endpoints - Set up Alembic for database migrations - Add comprehensive documentation in README.md
341 lines
11 KiB
Python
341 lines
11 KiB
Python
from typing import Any, Dict, List, Optional
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import func, and_, desc
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_db, get_current_active_user, get_current_admin_user
|
|
from app.models.user import User as UserModel
|
|
from app.models.product import Product
|
|
from app.models.inventory import Inventory, InventoryTransaction
|
|
from app.models.purchase_order import PurchaseOrder, PurchaseOrderItem
|
|
from app.models.sale import Sale, SaleItem
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/inventory-value")
|
|
def inventory_value_report(
|
|
db: Session = Depends(get_db),
|
|
current_user: UserModel = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get current inventory value report.
|
|
Calculates total inventory value (quantity * cost price).
|
|
"""
|
|
# Join Product and Inventory to calculate value
|
|
inventory_value = db.query(
|
|
func.sum(Product.cost_price * Inventory.quantity).label("total_value"),
|
|
func.count(Inventory.id).label("items_count"),
|
|
func.sum(Inventory.quantity).label("total_quantity")
|
|
).join(
|
|
Product, Product.id == Inventory.product_id
|
|
).filter(
|
|
Inventory.quantity > 0
|
|
).first()
|
|
|
|
# Get top products by value
|
|
top_products = db.query(
|
|
Product.id,
|
|
Product.name,
|
|
Product.sku,
|
|
Product.cost_price,
|
|
Inventory.quantity,
|
|
(Product.cost_price * Inventory.quantity).label("value")
|
|
).join(
|
|
Inventory, Product.id == Inventory.product_id
|
|
).filter(
|
|
Inventory.quantity > 0
|
|
).order_by(
|
|
desc("value")
|
|
).limit(10).all()
|
|
|
|
top_products_result = []
|
|
for p in top_products:
|
|
top_products_result.append({
|
|
"id": p.id,
|
|
"name": p.name,
|
|
"sku": p.sku,
|
|
"cost_price": float(p.cost_price) if p.cost_price else 0,
|
|
"quantity": p.quantity,
|
|
"value": float(p.value) if p.value else 0
|
|
})
|
|
|
|
return {
|
|
"total_value": float(inventory_value.total_value) if inventory_value.total_value else 0,
|
|
"items_count": inventory_value.items_count or 0,
|
|
"total_quantity": inventory_value.total_quantity or 0,
|
|
"top_products_by_value": top_products_result,
|
|
"report_date": datetime.now()
|
|
}
|
|
|
|
|
|
@router.get("/low-stock")
|
|
def low_stock_report(
|
|
db: Session = Depends(get_db),
|
|
threshold: int = 5,
|
|
current_user: UserModel = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get low stock report.
|
|
Shows products with inventory below specified threshold.
|
|
"""
|
|
# Get products with low stock
|
|
low_stock_products = db.query(
|
|
Product.id,
|
|
Product.name,
|
|
Product.sku,
|
|
Product.barcode,
|
|
func.sum(Inventory.quantity).label("total_quantity")
|
|
).outerjoin(
|
|
Inventory, Product.id == Inventory.product_id
|
|
).group_by(
|
|
Product.id
|
|
).having(
|
|
func.coalesce(func.sum(Inventory.quantity), 0) < threshold
|
|
).all()
|
|
|
|
result = []
|
|
for p in low_stock_products:
|
|
result.append({
|
|
"id": p.id,
|
|
"name": p.name,
|
|
"sku": p.sku,
|
|
"barcode": p.barcode,
|
|
"quantity": p.total_quantity or 0
|
|
})
|
|
|
|
return {
|
|
"threshold": threshold,
|
|
"count": len(result),
|
|
"products": result,
|
|
"report_date": datetime.now()
|
|
}
|
|
|
|
|
|
@router.get("/sales-summary")
|
|
def sales_summary_report(
|
|
db: Session = Depends(get_db),
|
|
start_date: Optional[datetime] = None,
|
|
end_date: Optional[datetime] = None,
|
|
current_user: UserModel = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get sales summary report for a specified date range.
|
|
If no dates provided, defaults to the last 30 days.
|
|
"""
|
|
# Set default date range if not provided
|
|
if not end_date:
|
|
end_date = datetime.now()
|
|
if not start_date:
|
|
start_date = end_date - timedelta(days=30)
|
|
|
|
# Get summary of all sales in date range
|
|
sales_summary = db.query(
|
|
func.count(Sale.id).label("total_sales"),
|
|
func.sum(SaleItem.quantity * SaleItem.unit_price).label("total_revenue"),
|
|
func.sum(SaleItem.quantity).label("total_items_sold")
|
|
).join(
|
|
SaleItem, Sale.id == SaleItem.sale_id
|
|
).filter(
|
|
Sale.status == "completed",
|
|
Sale.created_at >= start_date,
|
|
Sale.created_at <= end_date
|
|
).first()
|
|
|
|
# Get top selling products
|
|
top_products = db.query(
|
|
Product.id,
|
|
Product.name,
|
|
Product.sku,
|
|
func.sum(SaleItem.quantity).label("quantity_sold"),
|
|
func.sum(SaleItem.quantity * SaleItem.unit_price).label("revenue")
|
|
).join(
|
|
SaleItem, Product.id == SaleItem.product_id
|
|
).join(
|
|
Sale, SaleItem.sale_id == Sale.id
|
|
).filter(
|
|
Sale.status == "completed",
|
|
Sale.created_at >= start_date,
|
|
Sale.created_at <= end_date
|
|
).group_by(
|
|
Product.id
|
|
).order_by(
|
|
desc("quantity_sold")
|
|
).limit(10).all()
|
|
|
|
top_products_result = []
|
|
for p in top_products:
|
|
top_products_result.append({
|
|
"id": p.id,
|
|
"name": p.name,
|
|
"sku": p.sku,
|
|
"quantity_sold": p.quantity_sold,
|
|
"revenue": float(p.revenue) if p.revenue else 0
|
|
})
|
|
|
|
return {
|
|
"start_date": start_date,
|
|
"end_date": end_date,
|
|
"total_sales": sales_summary.total_sales or 0,
|
|
"total_revenue": float(sales_summary.total_revenue) if sales_summary.total_revenue else 0,
|
|
"total_items_sold": sales_summary.total_items_sold or 0,
|
|
"top_selling_products": top_products_result,
|
|
"report_date": datetime.now()
|
|
}
|
|
|
|
|
|
@router.get("/purchases-summary")
|
|
def purchases_summary_report(
|
|
db: Session = Depends(get_db),
|
|
start_date: Optional[datetime] = None,
|
|
end_date: Optional[datetime] = None,
|
|
current_user: UserModel = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get purchases summary report for a specified date range.
|
|
If no dates provided, defaults to the last 30 days.
|
|
"""
|
|
# Set default date range if not provided
|
|
if not end_date:
|
|
end_date = datetime.now()
|
|
if not start_date:
|
|
start_date = end_date - timedelta(days=30)
|
|
|
|
# Get summary of all received purchase orders in date range
|
|
purchases_summary = db.query(
|
|
func.count(PurchaseOrder.id).label("total_purchase_orders"),
|
|
func.sum(PurchaseOrderItem.quantity * PurchaseOrderItem.unit_price).label("total_cost"),
|
|
func.sum(PurchaseOrderItem.quantity).label("total_items_purchased")
|
|
).join(
|
|
PurchaseOrderItem, PurchaseOrder.id == PurchaseOrderItem.purchase_order_id
|
|
).filter(
|
|
PurchaseOrder.status == "received",
|
|
PurchaseOrder.created_at >= start_date,
|
|
PurchaseOrder.created_at <= end_date
|
|
).first()
|
|
|
|
# Get top suppliers
|
|
top_suppliers = db.query(
|
|
PurchaseOrder.supplier_name,
|
|
func.count(PurchaseOrder.id).label("order_count"),
|
|
func.sum(PurchaseOrderItem.quantity * PurchaseOrderItem.unit_price).label("total_spend")
|
|
).join(
|
|
PurchaseOrderItem, PurchaseOrder.id == PurchaseOrderItem.purchase_order_id
|
|
).filter(
|
|
PurchaseOrder.status == "received",
|
|
PurchaseOrder.created_at >= start_date,
|
|
PurchaseOrder.created_at <= end_date
|
|
).group_by(
|
|
PurchaseOrder.supplier_name
|
|
).order_by(
|
|
desc("total_spend")
|
|
).limit(5).all()
|
|
|
|
top_suppliers_result = []
|
|
for s in top_suppliers:
|
|
top_suppliers_result.append({
|
|
"supplier_name": s.supplier_name,
|
|
"order_count": s.order_count,
|
|
"total_spend": float(s.total_spend) if s.total_spend else 0
|
|
})
|
|
|
|
return {
|
|
"start_date": start_date,
|
|
"end_date": end_date,
|
|
"total_purchase_orders": purchases_summary.total_purchase_orders or 0,
|
|
"total_cost": float(purchases_summary.total_cost) if purchases_summary.total_cost else 0,
|
|
"total_items_purchased": purchases_summary.total_items_purchased or 0,
|
|
"top_suppliers": top_suppliers_result,
|
|
"report_date": datetime.now()
|
|
}
|
|
|
|
|
|
@router.get("/inventory-movements")
|
|
def inventory_movements_report(
|
|
db: Session = Depends(get_db),
|
|
product_id: Optional[int] = None,
|
|
start_date: Optional[datetime] = None,
|
|
end_date: Optional[datetime] = None,
|
|
current_user: UserModel = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get inventory movements report for a specified date range and product.
|
|
If no dates provided, defaults to the last 30 days.
|
|
"""
|
|
# Set default date range if not provided
|
|
if not end_date:
|
|
end_date = datetime.now()
|
|
if not start_date:
|
|
start_date = end_date - timedelta(days=30)
|
|
|
|
# Build query for inventory transactions
|
|
query = db.query(
|
|
InventoryTransaction.id,
|
|
InventoryTransaction.product_id,
|
|
Product.name.label("product_name"),
|
|
Product.sku,
|
|
InventoryTransaction.quantity,
|
|
InventoryTransaction.transaction_type,
|
|
InventoryTransaction.reference_id,
|
|
InventoryTransaction.reason,
|
|
InventoryTransaction.timestamp,
|
|
InventoryTransaction.location
|
|
).join(
|
|
Product, InventoryTransaction.product_id == Product.id
|
|
).filter(
|
|
InventoryTransaction.timestamp >= start_date,
|
|
InventoryTransaction.timestamp <= end_date
|
|
)
|
|
|
|
# Filter by product if specified
|
|
if product_id:
|
|
query = query.filter(InventoryTransaction.product_id == product_id)
|
|
|
|
# Execute query
|
|
transactions = query.order_by(InventoryTransaction.timestamp.desc()).all()
|
|
|
|
result = []
|
|
for t in transactions:
|
|
result.append({
|
|
"id": t.id,
|
|
"product_id": t.product_id,
|
|
"product_name": t.product_name,
|
|
"sku": t.sku,
|
|
"quantity": t.quantity,
|
|
"transaction_type": t.transaction_type,
|
|
"reference_id": t.reference_id,
|
|
"reason": t.reason,
|
|
"timestamp": t.timestamp,
|
|
"location": t.location
|
|
})
|
|
|
|
# Get summary by transaction type
|
|
summary = db.query(
|
|
InventoryTransaction.transaction_type,
|
|
func.sum(InventoryTransaction.quantity).label("total_quantity")
|
|
).filter(
|
|
InventoryTransaction.timestamp >= start_date,
|
|
InventoryTransaction.timestamp <= end_date
|
|
)
|
|
|
|
if product_id:
|
|
summary = summary.filter(InventoryTransaction.product_id == product_id)
|
|
|
|
summary = summary.group_by(InventoryTransaction.transaction_type).all()
|
|
|
|
summary_result = {}
|
|
for s in summary:
|
|
summary_result[s.transaction_type] = s.total_quantity
|
|
|
|
return {
|
|
"start_date": start_date,
|
|
"end_date": end_date,
|
|
"product_id": product_id,
|
|
"transaction_count": len(result),
|
|
"summary_by_type": summary_result,
|
|
"transactions": result,
|
|
"report_date": datetime.now()
|
|
} |