Add Mago4 demo ERP with Python FastAPI
Implements a working in-memory ERP demo inspired by Mago4, covering the core modules: dashboard with KPI widgets, customers, suppliers, products, sales orders, purchase orders, and warehouse inventory. - FastAPI backend with Pydantic models and REST API (auto-docs at /docs) - Jinja2 HTML templates with Mago4-style UI (dark sidebar, blue/orange/ green/red palette, Material Design icons) - In-memory fake database seeded with sample customers, suppliers, products, and orders — ready to swap for SQLAlchemy + SQL Server - Workarounds for Python 3.14 + Starlette 1.3 / Jinja2 compatibility (cache_size=0, new TemplateResponse(request, name, ctx) signature) - launch.json for one-click preview; run.bat for manual startup Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
@@ -0,0 +1,304 @@
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from fastapi.responses import HTMLResponse
|
||||
from starlette.requests import Request
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
from datetime import datetime
|
||||
import os
|
||||
|
||||
from models import (
|
||||
Customer, Supplier, Product, SalesOrder, PurchaseOrder,
|
||||
OrderLine, OrderStatus, Dashboard, DashboardWidget
|
||||
)
|
||||
from database import db
|
||||
|
||||
app = FastAPI(
|
||||
title="Mago4 Demo",
|
||||
description="Demo gestionale ERP con Python FastAPI",
|
||||
version="1.0.0"
|
||||
)
|
||||
|
||||
# Setup static e template files
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
os.makedirs(os.path.join(BASE_DIR, "static"), exist_ok=True)
|
||||
os.makedirs(os.path.join(BASE_DIR, "static", "css"), exist_ok=True)
|
||||
os.makedirs(os.path.join(BASE_DIR, "static", "js"), exist_ok=True)
|
||||
os.makedirs(os.path.join(BASE_DIR, "templates"), exist_ok=True)
|
||||
|
||||
app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static")
|
||||
|
||||
# Workaround Python 3.14 + Jinja2: disable template cache (globals dict not hashable as cache key)
|
||||
_jinja_env = Environment(
|
||||
loader=FileSystemLoader(os.path.join(BASE_DIR, "templates")),
|
||||
autoescape=True,
|
||||
cache_size=0,
|
||||
)
|
||||
templates = Jinja2Templates(env=_jinja_env)
|
||||
|
||||
|
||||
# ==================== DASHBOARD ====================
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
async def dashboard(request: Request):
|
||||
stats = db.get_dashboard_stats()
|
||||
|
||||
widgets = [
|
||||
[
|
||||
DashboardWidget(
|
||||
title="Ordini Vendita da Evadere",
|
||||
style="stats",
|
||||
size="small",
|
||||
icon="shopping_cart",
|
||||
color="blue",
|
||||
value=str(stats["sales_to_fulfill"])
|
||||
),
|
||||
DashboardWidget(
|
||||
title="Ordini Acquisto da Ricevere",
|
||||
style="stats",
|
||||
size="small",
|
||||
icon="local_shipping",
|
||||
color="orange",
|
||||
value=str(stats["purchase_to_receive"])
|
||||
),
|
||||
DashboardWidget(
|
||||
title="Crediti Clienti",
|
||||
style="stats",
|
||||
size="small",
|
||||
icon="euro_symbol",
|
||||
color="green",
|
||||
value=f"€ {stats['total_receivable']:,.2f}",
|
||||
format="money"
|
||||
),
|
||||
DashboardWidget(
|
||||
title="Articoli Sotto Soglia",
|
||||
style="stats",
|
||||
size="small",
|
||||
icon="warning",
|
||||
color="red",
|
||||
value=str(stats["low_stock_count"])
|
||||
),
|
||||
],
|
||||
[
|
||||
DashboardWidget(
|
||||
title="Ultimi Ordini Vendita",
|
||||
subtitle="per data ordine",
|
||||
style="grid",
|
||||
size="medium",
|
||||
data={"orders": [
|
||||
{
|
||||
"order_number": o.order_number,
|
||||
"order_date": o.order_date.strftime("%d/%m/%Y"),
|
||||
"customer_name": o.customer_name,
|
||||
"total_amount": f"€ {o.total_amount:,.2f}",
|
||||
"status": o.status.value
|
||||
}
|
||||
for o in sorted(db.get_sales_orders(), key=lambda x: x.order_date, reverse=True)[:5]
|
||||
]}
|
||||
),
|
||||
DashboardWidget(
|
||||
title="Ultimi Ordini Acquisto",
|
||||
subtitle="per data ordine",
|
||||
style="grid",
|
||||
size="medium",
|
||||
data={"orders": [
|
||||
{
|
||||
"order_number": o.order_number,
|
||||
"order_date": o.order_date.strftime("%d/%m/%Y"),
|
||||
"supplier_name": o.supplier_name,
|
||||
"total_amount": f"€ {o.total_amount:,.2f}",
|
||||
"status": o.status.value
|
||||
}
|
||||
for o in sorted(db.get_purchase_orders(), key=lambda x: x.order_date, reverse=True)[:5]
|
||||
]}
|
||||
),
|
||||
]
|
||||
]
|
||||
|
||||
return templates.TemplateResponse(request, "dashboard.html", {
|
||||
"widgets": widgets,
|
||||
"stats": stats
|
||||
})
|
||||
|
||||
|
||||
# ==================== CUSTOMERS ====================
|
||||
@app.get("/api/customers", tags=["Customers"])
|
||||
async def list_customers():
|
||||
return db.get_customers()
|
||||
|
||||
|
||||
@app.get("/api/customers/{customer_id}", tags=["Customers"])
|
||||
async def get_customer(customer_id: int):
|
||||
customer = db.get_customer(customer_id)
|
||||
if not customer:
|
||||
raise HTTPException(status_code=404, detail="Cliente non trovato")
|
||||
return customer
|
||||
|
||||
|
||||
@app.post("/api/customers", tags=["Customers"])
|
||||
async def create_customer(customer: Customer):
|
||||
return db.create_customer(customer)
|
||||
|
||||
|
||||
@app.put("/api/customers/{customer_id}", tags=["Customers"])
|
||||
async def update_customer(customer_id: int, customer_data: dict):
|
||||
result = db.update_customer(customer_id, customer_data)
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="Cliente non trovato")
|
||||
return result
|
||||
|
||||
|
||||
@app.get("/customers", response_class=HTMLResponse)
|
||||
async def customers_page(request: Request):
|
||||
return templates.TemplateResponse(request, "customers.html", {
|
||||
"customers": db.get_customers()
|
||||
})
|
||||
|
||||
|
||||
# ==================== SUPPLIERS ====================
|
||||
@app.get("/api/suppliers", tags=["Suppliers"])
|
||||
async def list_suppliers():
|
||||
return db.get_suppliers()
|
||||
|
||||
|
||||
@app.get("/api/suppliers/{supplier_id}", tags=["Suppliers"])
|
||||
async def get_supplier(supplier_id: int):
|
||||
supplier = db.get_supplier(supplier_id)
|
||||
if not supplier:
|
||||
raise HTTPException(status_code=404, detail="Fornitore non trovato")
|
||||
return supplier
|
||||
|
||||
|
||||
@app.post("/api/suppliers", tags=["Suppliers"])
|
||||
async def create_supplier(supplier: Supplier):
|
||||
return db.create_supplier(supplier)
|
||||
|
||||
|
||||
@app.get("/suppliers", response_class=HTMLResponse)
|
||||
async def suppliers_page(request: Request):
|
||||
return templates.TemplateResponse(request, "suppliers.html", {
|
||||
"suppliers": db.get_suppliers()
|
||||
})
|
||||
|
||||
|
||||
# ==================== PRODUCTS ====================
|
||||
@app.get("/api/products", tags=["Products"])
|
||||
async def list_products():
|
||||
return db.get_products()
|
||||
|
||||
|
||||
@app.get("/api/products/{product_id}", tags=["Products"])
|
||||
async def get_product(product_id: int):
|
||||
product = db.get_product(product_id)
|
||||
if not product:
|
||||
raise HTTPException(status_code=404, detail="Articolo non trovato")
|
||||
return product
|
||||
|
||||
|
||||
@app.post("/api/products", tags=["Products"])
|
||||
async def create_product(product: Product):
|
||||
return db.create_product(product)
|
||||
|
||||
|
||||
@app.get("/products", response_class=HTMLResponse)
|
||||
async def products_page(request: Request):
|
||||
products = db.get_products()
|
||||
return templates.TemplateResponse(request, "products.html", {
|
||||
"products": products,
|
||||
"low_stock": [p for p in products if p.quantity_in_stock <= p.reorder_level]
|
||||
})
|
||||
|
||||
|
||||
# ==================== SALES ORDERS ====================
|
||||
@app.get("/api/sales-orders", tags=["Sales Orders"])
|
||||
async def list_sales_orders():
|
||||
return db.get_sales_orders()
|
||||
|
||||
|
||||
@app.get("/api/sales-orders/{order_id}", tags=["Sales Orders"])
|
||||
async def get_sales_order(order_id: int):
|
||||
order = db.get_sales_order(order_id)
|
||||
if not order:
|
||||
raise HTTPException(status_code=404, detail="Ordine vendita non trovato")
|
||||
return order
|
||||
|
||||
|
||||
@app.post("/api/sales-orders", tags=["Sales Orders"])
|
||||
async def create_sales_order(order: SalesOrder):
|
||||
return db.create_sales_order(order)
|
||||
|
||||
|
||||
@app.get("/sales-orders", response_class=HTMLResponse)
|
||||
async def sales_orders_page(request: Request):
|
||||
orders = db.get_sales_orders()
|
||||
return templates.TemplateResponse(request, "sales_orders.html", {
|
||||
"orders": orders,
|
||||
"pending": [o for o in orders if o.status != OrderStatus.RECEIVED]
|
||||
})
|
||||
|
||||
|
||||
# ==================== PURCHASE ORDERS ====================
|
||||
@app.get("/api/purchase-orders", tags=["Purchase Orders"])
|
||||
async def list_purchase_orders():
|
||||
return db.get_purchase_orders()
|
||||
|
||||
|
||||
@app.get("/api/purchase-orders/{order_id}", tags=["Purchase Orders"])
|
||||
async def get_purchase_order(order_id: int):
|
||||
order = db.get_purchase_order(order_id)
|
||||
if not order:
|
||||
raise HTTPException(status_code=404, detail="Ordine acquisto non trovato")
|
||||
return order
|
||||
|
||||
|
||||
@app.post("/api/purchase-orders", tags=["Purchase Orders"])
|
||||
async def create_purchase_order(order: PurchaseOrder):
|
||||
return db.create_purchase_order(order)
|
||||
|
||||
|
||||
@app.get("/purchase-orders", response_class=HTMLResponse)
|
||||
async def purchase_orders_page(request: Request):
|
||||
orders = db.get_purchase_orders()
|
||||
return templates.TemplateResponse(request, "purchase_orders.html", {
|
||||
"orders": orders,
|
||||
"pending": [o for o in orders if o.status != OrderStatus.RECEIVED]
|
||||
})
|
||||
|
||||
|
||||
# ==================== WAREHOUSE ====================
|
||||
@app.get("/api/warehouse/stats", tags=["Warehouse"])
|
||||
async def warehouse_stats():
|
||||
stats = db.get_dashboard_stats()
|
||||
return {
|
||||
"total_products": stats["total_products"],
|
||||
"low_stock_count": stats["low_stock_count"],
|
||||
"low_stock_items": stats["low_stock_items"],
|
||||
"all_products": db.get_products()
|
||||
}
|
||||
|
||||
|
||||
@app.get("/warehouse", response_class=HTMLResponse)
|
||||
async def warehouse_page(request: Request):
|
||||
stats = db.get_dashboard_stats()
|
||||
return templates.TemplateResponse(request, "warehouse.html", {
|
||||
"products": db.get_products(),
|
||||
"low_stock": stats["low_stock_items"]
|
||||
})
|
||||
|
||||
|
||||
# ==================== INFO ====================
|
||||
@app.get("/api/info", tags=["Info"])
|
||||
async def api_info():
|
||||
stats = db.get_dashboard_stats()
|
||||
return {
|
||||
"name": "Mago4 Demo",
|
||||
"version": "1.0.0",
|
||||
"status": "active",
|
||||
"database_type": "in-memory",
|
||||
"statistics": stats
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
port = int(os.environ.get("PORT", 8000))
|
||||
uvicorn.run(app, host="127.0.0.1", port=port)
|
||||
Reference in New Issue
Block a user