Merge pull request #550 from fastapiadmin/dev

feat(P2-1): SQL 控制台（仅外部 MySQL/PostgreSQL）

Author: fastapiadmin

backend/app/api/v1/routers.py

@@ -4,6 +4,7 @@
 
 from app.modules.ai.chat.controller import ChatRouter
 from app.modules.common.file.controller import FileRouter
+from app.modules.dbconsole.controller import DbConsoleRouter
 from app.modules.generator.gencode.controller import GenRouter
 from app.modules.monitor.cache.controller import CacheRouter
 from app.modules.monitor.health.controller import HealthRouter
@@ -61,6 +62,7 @@
     "/ai": [ChatRouter],
     "/generator": [GenRouter],
     "/common": [FileRouter],
+    "/dbconsole": [DbConsoleRouter],
 }
 
 api_v1 = APIRouter()

backend/app/modules/dbconsole/__init__.py

@@ -0,0 +1,127 @@
+"""SQL 控制台接口（仅外部数据库连接）。"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
+from fastapi.responses import JSONResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.common.response import ResponseSchema, SuccessResponse
+from app.core.base_schema import AuthSchema
+from app.core.dependencies import AuthPermission, db_getter
+from app.core.router_class import OperationLogRoute
+
+from .schema import (
+    DbConnectionCreateSchema,
+    DbConnectionOutSchema,
+    DbConnectionTestSchema,
+    DbConnectionUpdateSchema,
+    DbObjectSchema,
+    DbQueryResultSchema,
+    DbQuerySchema,
+)
+from .service import DbConsoleService
+
+DbConsoleRouter = APIRouter(route_class=OperationLogRoute, prefix="/connection", tags=["SQL控制台"])
+
+
+@DbConsoleRouter.get("/list", summary="查询数据库连接列表", response_model=ResponseSchema[list[DbConnectionOutSchema]])
+async def list_db_connection_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:query"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).get_list()
+    return SuccessResponse(data=result, msg="查询连接列表成功")
+
+
+@DbConsoleRouter.get("/detail/{id}", summary="查询数据库连接详情", response_model=ResponseSchema[DbConnectionOutSchema])
+async def detail_db_connection_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:query"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="连接ID", ge=1)],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).detail(id=id)
+    return SuccessResponse(data=result, msg="查询连接详情成功")
+
+
+@DbConsoleRouter.post("/create", status_code=status.HTTP_201_CREATED, summary="新增数据库连接", response_model=ResponseSchema[DbConnectionOutSchema])
+async def create_db_connection_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:create"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[DbConnectionCreateSchema, Body(description="连接配置")],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).create(data=data)
+    return SuccessResponse(data=result, msg="新增连接成功")
+
+
+@DbConsoleRouter.put("/update/{id}", summary="修改数据库连接", response_model=ResponseSchema[DbConnectionOutSchema])
+async def update_db_connection_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:update"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="连接ID", ge=1)],
+    data: Annotated[DbConnectionUpdateSchema, Body(description="连接配置")],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).update(id=id, data=data)
+    return SuccessResponse(data=result, msg="修改连接成功")
+
+
+@DbConsoleRouter.delete("/delete", summary="删除数据库连接", response_model=ResponseSchema[None])
+async def delete_db_connection_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:delete"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    ids: Annotated[list[int], Body(description="连接ID列表")],
+) -> JSONResponse:
+    await DbConsoleService(auth, db).delete(ids=ids)
+    return SuccessResponse(msg="删除连接成功")
+
+
+@DbConsoleRouter.post("/test", summary="按填写中的凭据测试连接（不落库）", response_model=ResponseSchema[bool])
+async def test_db_connection_config_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:update"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[DbConnectionTestSchema, Body(description="连接配置")],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).test_connection(data)
+    return SuccessResponse(data=result, msg="连接成功")
+
+
+@DbConsoleRouter.post("/test/{id}", summary="测试已保存的连接", response_model=ResponseSchema[bool])
+async def test_db_connection_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:query"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="连接ID", ge=1)],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).test_by_id(id=id)
+    return SuccessResponse(data=result, msg="连接成功")
+
+
+@DbConsoleRouter.get("/{id}/databases", summary="查询库/schema 列表", response_model=ResponseSchema[list[DbObjectSchema]])
+async def list_db_databases_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:query:execute"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="连接ID", ge=1)],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).list_databases(id=id)
+    return SuccessResponse(data=result, msg="查询库列表成功")
+
+
+@DbConsoleRouter.get("/{id}/tables", summary="查询表列表", response_model=ResponseSchema[list[DbObjectSchema]])
+async def list_db_tables_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:query:execute"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="连接ID", ge=1)],
+    database: Annotated[str | None, Query(max_length=128, description="库/schema 名")] = None,
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).list_tables(id=id, database=database)
+    return SuccessResponse(data=result, msg="查询表列表成功")
+
+
+@DbConsoleRouter.post("/{id}/query", summary="执行 SQL（只读校验；写需 confirmed）", response_model=ResponseSchema[DbQueryResultSchema])
+async def run_db_query_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:query:execute"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="连接ID", ge=1)],
+    data: Annotated[DbQuerySchema, Body(description="SQL 与确认标记")],
+) -> JSONResponse:
+    result = await DbConsoleService(auth, db).run_query(id=id, sql=data.sql, confirmed=data.confirmed)
+    return SuccessResponse(data=result, msg="执行成功")

backend/app/modules/dbconsole/controller.py

@@ -0,0 +1,14 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.base_crud import CRUDBase
+from app.core.base_schema import AuthSchema
+
+from .model import DbConnectionModel
+from .schema import DbConnectionCreateSchema, DbConnectionUpdateSchema
+
+
+class DbConnectionCRUD(CRUDBase[DbConnectionModel, DbConnectionCreateSchema, DbConnectionUpdateSchema]):
+    """SQL 控制台连接数据层"""
+
+    def __init__(self, auth: AuthSchema, db: AsyncSession) -> None:
+        super().__init__(model=DbConnectionModel, auth=auth, db=db)

backend/app/modules/dbconsole/crud.py

@@ -0,0 +1,27 @@
+"""SQL 控制台：**外部**数据库连接（仅 mysql / postgres）。
+
+按需求**不包含**本地/内置数据库（例如 FastHub 的 `data/fasthub.db` 这类本机 SQLite 浏览）。
+密码用 Fernet 加密存储（CryptoUtil），出参永不返回密码，只回 `has_password`。
+"""
+
+from sqlalchemy import Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.base_model import ModelMixin, UserMixin
+
+
+class DbConnectionModel(ModelMixin, UserMixin):
+    """外部数据库连接配置"""
+
+    __tablename__: str = "dbconsole_connection"
+    __table_args__: dict[str, str] = {"comment": "SQL控制台外部数据库连接表"}
+
+    name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True, comment="连接名称")
+    db_type: Mapped[str] = mapped_column(String(16), nullable=False, index=True, comment="数据库类型(mysql/postgres)")
+    host: Mapped[str] = mapped_column(String(255), nullable=False, comment="主机地址")
+    port: Mapped[int] = mapped_column(Integer, nullable=False, comment="端口")
+    username: Mapped[str] = mapped_column(String(128), nullable=False, comment="用户名")
+    password: Mapped[str] = mapped_column(Text, nullable=False, comment="密码(Fernet加密)")
+    database: Mapped[str | None] = mapped_column(String(128), default=None, nullable=True, comment="默认库名")
+    status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启用 1:停用)")
+    description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")

backend/app/modules/dbconsole/model.py

@@ -0,0 +1,76 @@
+"""SQL 控制台数据契约。"""
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class DbConnectionBaseSchema(BaseModel):
+    """连接配置公共字段（不含密码）"""
+
+    name: str = Field(..., min_length=1, max_length=64, description="连接名称（唯一）")
+    db_type: Literal["mysql", "postgres"] = Field(..., description="数据库类型（仅外部连接：mysql/postgres）")
+    host: str = Field(..., min_length=1, max_length=255, description="主机地址")
+    port: int = Field(..., ge=1, le=65535, description="端口")
+    username: str = Field(..., min_length=1, max_length=128, description="用户名")
+    database: str | None = Field(None, max_length=128, description="默认库名（可空）")
+    description: str | None = Field(None, max_length=500, description="备注")
+
+
+class DbConnectionCreateSchema(DbConnectionBaseSchema):
+    """新建连接"""
+
+    password: str = Field(..., min_length=1, max_length=500, description="密码（加密存储）")
+
+
+class DbConnectionUpdateSchema(DbConnectionBaseSchema):
+    """修改连接（不传密码表示不修改）"""
+
+    password: str | None = Field(None, max_length=500, description="密码；不传表示不修改")
+
+
+class DbConnectionTestSchema(BaseModel):
+    """按填写中的凭据测试连接（不落库）"""
+
+    db_type: Literal["mysql", "postgres"] = Field(..., description="数据库类型")
+    host: str = Field(..., min_length=1, max_length=255)
+    port: int = Field(..., ge=1, le=65535)
+    username: str = Field(..., min_length=1, max_length=128)
+    password: str = Field(..., min_length=1, max_length=500)
+    database: str | None = Field(None, max_length=128)
+
+
+class DbConnectionOutSchema(DbConnectionBaseSchema):
+    """连接出参（不含密码）"""
+
+    id: int = Field(..., description="连接ID")
+    status: int = Field(..., description="状态(0:启用 1:停用)")
+    has_password: bool = Field(False, description="是否已设置密码")
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class DbQuerySchema(BaseModel):
+    """执行 SQL"""
+
+    sql: str = Field(..., min_length=1, max_length=20000, description="SQL 语句")
+    confirmed: bool = Field(False, description="写操作必须显式确认（true）")
+
+
+class DbQueryResultSchema(BaseModel):
+    """SQL 执行结果"""
+
+    sql: str = Field(..., description="实际执行的 SQL")
+    read_only: bool = Field(..., description="是否为只读语句")
+    columns: list[str] = Field(default_factory=list, description="列名")
+    rows: list[list[Any]] = Field(default_factory=list, description="行（最多 200 行）")
+    rowcount: int = Field(0, description="影响行数/返回行数")
+    masked_columns: list[str] = Field(default_factory=list, description="已脱敏的敏感列")
+    truncated: bool = Field(False, description="是否因行数上限被截断")
+    elapsed_ms: int = Field(0, description="耗时（毫秒）")
+
+
+class DbObjectSchema(BaseModel):
+    """库/表名"""
+
+    name: str = Field(..., description="名称")