Codex
190 lines
7.0 KiB
Python
190 lines
7.0 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""数据库连接管理 API"""
|
|
import logging
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.base_schema import PaginatedResponse, ResponseModel
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from core.application.model import Application
|
|
from core.database_connection.schema import (
|
|
DatabaseConnectionCreate,
|
|
DatabaseConnectionResponse,
|
|
DatabaseConnectionSimpleOut,
|
|
DatabaseConnectionTestRequest,
|
|
DatabaseConnectionTestResponse,
|
|
DatabaseConnectionUpdate,
|
|
)
|
|
from core.database_connection.service import DatabaseConnectionService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/database-connection", tags=["数据库连接管理"])
|
|
|
|
|
|
@router.get("", response_model=PaginatedResponse[DatabaseConnectionResponse], summary="获取连接列表")
|
|
async def list_connections(
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=settings.PAGE_SIZE, ge=1, le=settings.PAGE_MAX_SIZE, alias="pageSize"),
|
|
application_id: Optional[str] = Query(default=None, alias="applicationId"),
|
|
name: Optional[str] = None,
|
|
code: Optional[str] = None,
|
|
status: Optional[bool] = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
items, total = await DatabaseConnectionService.get_list(
|
|
db,
|
|
page=page,
|
|
page_size=page_size,
|
|
application_id=application_id,
|
|
name=name,
|
|
code=code,
|
|
status=status,
|
|
)
|
|
|
|
app_ids = list({item.application_id for item in items if item.application_id})
|
|
app_name_map = {}
|
|
if app_ids:
|
|
app_result = await db.execute(
|
|
select(Application.id, Application.name).where(Application.id.in_(app_ids))
|
|
)
|
|
app_name_map = {row.id: row.name for row in app_result}
|
|
|
|
result_items = []
|
|
for item in items:
|
|
item_dict = DatabaseConnectionService.to_response_dict(item)
|
|
item_dict["application_name"] = app_name_map.get(item.application_id, "")
|
|
result_items.append(item_dict)
|
|
|
|
return PaginatedResponse(items=result_items, total=total)
|
|
|
|
|
|
@router.post("", response_model=DatabaseConnectionResponse, summary="创建连接")
|
|
async def create_connection(
|
|
data: DatabaseConnectionCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
try:
|
|
DatabaseConnectionService.validate_code(data.code)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
if await DatabaseConnectionService.check_code_exists(db, data.code):
|
|
raise HTTPException(status_code=400, detail=f"编码已存在: {data.code}")
|
|
|
|
try:
|
|
row = await DatabaseConnectionService.create(db, data.model_dump())
|
|
await db.commit()
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
return DatabaseConnectionService.to_response_dict(row)
|
|
|
|
|
|
@router.get("/get/all", response_model=List[DatabaseConnectionSimpleOut], summary="获取全部启用连接")
|
|
async def get_all_connections(
|
|
application_id: Optional[str] = Query(default=None, alias="applicationId"),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
configs = await DatabaseConnectionService.get_manager_configs(db, application_id)
|
|
return [
|
|
DatabaseConnectionSimpleOut(
|
|
code=item["db_name"],
|
|
name=item.get("display_name") or item["name"],
|
|
db_type=item["db_type"],
|
|
host=item["host"],
|
|
port=item["port"],
|
|
database=item["database"],
|
|
is_system=item.get("is_system", False),
|
|
)
|
|
for item in configs
|
|
]
|
|
|
|
|
|
@router.get("/check-code/{code}", summary="检查编码是否可用")
|
|
async def check_code_available(code: str, db: AsyncSession = Depends(get_db)):
|
|
if code == "default":
|
|
return {"available": False, "message": "default is reserved"}
|
|
exists = await DatabaseConnectionService.check_code_exists(db, code)
|
|
return {"available": not exists}
|
|
|
|
|
|
@router.post("/test", response_model=DatabaseConnectionTestResponse, summary="测试未保存连接")
|
|
async def test_unsaved_connection(data: DatabaseConnectionTestRequest):
|
|
try:
|
|
result = await DatabaseConnectionService.test_config(data.model_dump())
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
if not result.get("success"):
|
|
logger.error(
|
|
"Database connection test failed (unsaved): host=%s port=%s db_type=%s message=%s",
|
|
data.host,
|
|
data.port,
|
|
data.db_type,
|
|
result.get("message"),
|
|
)
|
|
return DatabaseConnectionTestResponse(**result)
|
|
|
|
|
|
@router.get("/{conn_id}", response_model=DatabaseConnectionResponse, summary="获取连接详情")
|
|
async def get_connection(conn_id: str, db: AsyncSession = Depends(get_db)):
|
|
row = await DatabaseConnectionService.get_by_id(db, conn_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="连接不存在")
|
|
return DatabaseConnectionService.to_response_dict(row)
|
|
|
|
|
|
@router.put("/{conn_id}", response_model=DatabaseConnectionResponse, summary="更新连接")
|
|
async def update_connection(
|
|
conn_id: str,
|
|
data: DatabaseConnectionUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
row = await DatabaseConnectionService.get_by_id(db, conn_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="连接不存在")
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
try:
|
|
row = await DatabaseConnectionService.update(db, row, update_data)
|
|
await db.commit()
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
return DatabaseConnectionService.to_response_dict(row)
|
|
|
|
|
|
@router.delete("/{conn_id}", response_model=ResponseModel, summary="删除连接")
|
|
async def delete_connection(conn_id: str, db: AsyncSession = Depends(get_db)):
|
|
row = await DatabaseConnectionService.get_by_id(db, conn_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="连接不存在")
|
|
try:
|
|
await DatabaseConnectionService.delete(db, row)
|
|
await db.commit()
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
return ResponseModel(message="删除成功")
|
|
|
|
|
|
@router.post("/{conn_id}/test", response_model=DatabaseConnectionTestResponse, summary="测试已保存连接")
|
|
async def test_saved_connection(conn_id: str, db: AsyncSession = Depends(get_db)):
|
|
row = await DatabaseConnectionService.get_by_id(db, conn_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="连接不存在")
|
|
result = await DatabaseConnectionService.test_saved(db, row)
|
|
if not result.get("success"):
|
|
logger.error(
|
|
"Database connection test failed (saved): id=%s code=%s name=%s message=%s",
|
|
conn_id,
|
|
row.code,
|
|
row.name,
|
|
result.get("message"),
|
|
)
|
|
return DatabaseConnectionTestResponse(**result)
|