51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from app.db.session import check_database_health, get_engine
|
|
from app.schemas.database import DataCountItem
|
|
from app.services.auth import require_authenticated_user
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/health")
|
|
def health() -> dict[str, object]:
|
|
try:
|
|
db_status = check_database_health()
|
|
except (SQLAlchemyError, RuntimeError, Exception) as exc:
|
|
return {
|
|
"app": "ok",
|
|
"database": {
|
|
"connected": False,
|
|
"error": str(exc),
|
|
},
|
|
}
|
|
|
|
return {
|
|
"app": "ok",
|
|
"database": db_status,
|
|
}
|
|
|
|
|
|
@router.get("/data-summary", response_model=list[DataCountItem], dependencies=[Depends(require_authenticated_user)])
|
|
def data_summary() -> list[DataCountItem]:
|
|
summary_tables = [
|
|
("dept_count", "部门数量", "sys_department"),
|
|
("warehouse_count", "仓库数量", "wh_warehouse"),
|
|
("item_count", "物料数量", "md_item"),
|
|
("product_count", "产品数量", "md_product"),
|
|
("customer_count", "客户数量", "md_customer"),
|
|
("sales_order_count", "订单数量", "so_sales_order"),
|
|
("mrp_count", "MRP需求数量", "mrp_material_demand"),
|
|
("work_order_count", "工单数量", "pp_work_order"),
|
|
]
|
|
|
|
engine = get_engine()
|
|
result: list[DataCountItem] = []
|
|
with engine.connect() as conn:
|
|
for code, label, table_name in summary_tables:
|
|
count = conn.execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar() or 0
|
|
result.append(DataCountItem(code=code, label=label, count=int(count)))
|
|
return result
|