#!/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)