Codex
107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
数据库监控 API
|
|
|
|
监控维度与「数据库连接管理」一致：每个启用连接（code）为一个监控目标，
|
|
凭证通过 ConnectionResolver 解析，支持 PostgreSQL / MySQL / SQL Server / Oracle。
|
|
"""
|
|
import logging
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from core.database_monitor.config_resolver import (
|
|
create_monitor_collector,
|
|
list_monitor_configs,
|
|
resolve_monitor_target,
|
|
)
|
|
from core.database_monitor.schema import (
|
|
DatabaseConfigSchema,
|
|
DatabaseConnectionTestSchema,
|
|
DatabaseOverviewSchema,
|
|
DatabaseRealtimeStatsSchema,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/database_monitor", tags=["数据库监控"])
|
|
|
|
|
|
async def get_database_configs(db: AsyncSession) -> List[dict]:
|
|
"""获取全部可监控连接配置（供 WebSocket 等模块复用）。"""
|
|
return await list_monitor_configs(db)
|
|
|
|
|
|
@router.get("/configs", response_model=List[DatabaseConfigSchema], summary="获取数据库监控目标列表")
|
|
async def get_database_monitor_configs(db: AsyncSession = Depends(get_db)):
|
|
"""获取已注册的数据库连接列表，作为监控目标。"""
|
|
configs = await list_monitor_configs(db)
|
|
return [
|
|
DatabaseConfigSchema(
|
|
name=config["name"],
|
|
db_name=config["db_name"],
|
|
db_type=config["db_type"],
|
|
host=config["host"],
|
|
port=config["port"],
|
|
database=config["database"],
|
|
user=config["user"],
|
|
has_password=config["has_password"],
|
|
is_system=config.get("is_system", False),
|
|
)
|
|
for config in configs
|
|
]
|
|
|
|
|
|
@router.get("/{connection_code}/overview", response_model=DatabaseOverviewSchema, summary="获取数据库概览信息")
|
|
async def get_database_overview(
|
|
connection_code: str,
|
|
database: Optional[str] = Query(None, description="覆盖连接默认库/服务名"),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""获取指定连接的监控概览。"""
|
|
try:
|
|
target = await resolve_monitor_target(db, connection_code, database)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
collector = create_monitor_collector(target)
|
|
data = await collector.get_all_info(connection_code, target["name"])
|
|
return DatabaseOverviewSchema(**data)
|
|
|
|
|
|
@router.get("/{connection_code}/realtime", response_model=DatabaseRealtimeStatsSchema, summary="获取数据库实时统计")
|
|
async def get_database_realtime_stats(
|
|
connection_code: str,
|
|
database: Optional[str] = Query(None, description="覆盖连接默认库/服务名"),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""获取指定连接的实时统计。"""
|
|
try:
|
|
target = await resolve_monitor_target(db, connection_code, database)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
collector = create_monitor_collector(target)
|
|
data = await collector.get_realtime_stats(connection_code)
|
|
return DatabaseRealtimeStatsSchema(**data)
|
|
|
|
|
|
@router.post("/{connection_code}/test", response_model=DatabaseConnectionTestSchema, summary="测试数据库连接")
|
|
async def test_database_connection(
|
|
connection_code: str,
|
|
database: Optional[str] = Query(None, description="覆盖连接默认库/服务名"),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""测试指定连接是否可用。"""
|
|
try:
|
|
target = await resolve_monitor_target(db, connection_code, database)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
collector = create_monitor_collector(target)
|
|
result = await collector.test_connection()
|
|
return DatabaseConnectionTestSchema(**result)
|