feat: Add Api Docs

Author: XiaJunjie2020

backend/apps/datasource/api/datasource.py

@@ -8,10 +8,11 @@
 
 import orjson
 import pandas as pd
-from fastapi import APIRouter, File, UploadFile, HTTPException
+from fastapi import APIRouter, File, UploadFile, HTTPException, Path
 
 from apps.db.db import get_schema
 from apps.db.engine import get_engine_conn
+from apps.swagger.i18n import PLACEHOLDER_PREFIX
 from common.core.config import settings
 from common.core.deps import SessionDep, CurrentUser, Trans
 from common.utils.utils import SQLBotLogUtil
@@ -22,7 +23,7 @@
 from ..crud.table import get_tables_by_ds_id
 from ..models.datasource import CoreDatasource, CreateDatasource, TableObj, CoreTable, CoreField, FieldObj
 
-router = APIRouter(tags=["datasource"], prefix="/datasource")
+router = APIRouter(tags=["Datasource"], prefix="/datasource")
 path = settings.EXCEL_PATH
 
 
@@ -33,13 +34,14 @@ async def query_by_oid(session: SessionDep, user: CurrentUser, oid: int) -> List
     return get_datasource_list(session=session, user=user, oid=oid)
 
 
-@router.get("/list")
+@router.get("/list", response_model=List[CoreDatasource], summary=f"{PLACEHOLDER_PREFIX}ds_list",
+            description=f"{PLACEHOLDER_PREFIX}ds_list_description")
 async def datasource_list(session: SessionDep, user: CurrentUser):
     return get_datasource_list(session=session, user=user)
 
 
-@router.post("/get/{id}")
-async def get_datasource(session: SessionDep, id: int):
+@router.post("/get/{id}", response_model=CoreDatasource, summary=f"{PLACEHOLDER_PREFIX}ds_get")
+async def get_datasource(session: SessionDep, id: int = Path(..., description=f"{PLACEHOLDER_PREFIX}ds_id")):
     return get_ds(session, id)
 
 

backend/apps/swagger/__init__.py

@@ -0,0 +1,2 @@
+# Author: Junjun
+# Date: 2025/12/11

backend/apps/swagger/i18n.py

@@ -0,0 +1,51 @@
+# Author: Junjun
+# Date: 2025/12/11
+# i18n.py
+import json
+from pathlib import Path
+from typing import Dict
+
+# placeholder prefix（trans key prefix）
+PLACEHOLDER_PREFIX = "PLACEHOLDER_"
+
+# default lang
+DEFAULT_LANG = "en"
+
+LOCALES_DIR = Path(__file__).parent / "locales"
+_translations_cache: Dict[str, Dict[str, str]] = {}
+
+
+def load_translation(lang: str) -> Dict[str, str]:
+    """Load translations for the specified language from a JSON file"""
+    if lang in _translations_cache:
+        return _translations_cache[lang]
+
+    file_path = LOCALES_DIR / f"{lang}.json"
+    if not file_path.exists():
+        if lang == DEFAULT_LANG:
+            raise FileNotFoundError(f"Default language file not found: {file_path}")
+        # If the non-default language is missing, fall back to the default language
+        return load_translation(DEFAULT_LANG)
+
+    try:
+        with open(file_path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+            if not isinstance(data, dict):
+                raise ValueError(f"Translation file {file_path} must be a JSON object")
+            _translations_cache[lang] = data
+            return data
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Invalid JSON in {file_path}: {e}")
+
+
+# group tags
+tags_metadata = [
+    {
+        "name": "Datasource",
+        "description": f"{PLACEHOLDER_PREFIX}ds_api"
+    }
+]
+
+
+def get_translation(lang: str) -> Dict[str, str]:
+    return load_translation(lang)

backend/apps/swagger/locales/en.json

@@ -0,0 +1,7 @@
+{
+  "ds_api": "Datasource API",
+  "ds_list": "Datasource list",
+  "ds_list_description": "Retrieve all data sources under the current workspace",
+  "ds_get": "Get Datasource",
+  "ds_id": "Datasource ID"
+}

backend/apps/swagger/locales/zh.json

@@ -0,0 +1,7 @@
+{
+  "ds_api": "数据源接口",
+  "ds_list": "数据源列表",
+  "ds_list_description": "获取当前工作空间下所有数据源",
+  "ds_get": "获取数据源",
+  "ds_id": "数据源 ID"
+}

backend/common/core/response_middleware.py

@@ -18,7 +18,10 @@ async def dispatch(self, request, call_next):
 
         direct_paths = [
             f"{settings.API_V1_STR}/mcp/mcp_question",
-            f"{settings.API_V1_STR}/mcp/mcp_assistant"
+            f"{settings.API_V1_STR}/mcp/mcp_assistant",
+            "/openapi.json",
+            "/docs",
+            "/redoc"
         ]
 
         route = request.scope.get("route")

backend/main.py

@@ -1,9 +1,12 @@
 import os
+from typing import Dict, Any
 
 import sqlbot_xpack
 from alembic.config import Config
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.concurrency import asynccontextmanager
+from fastapi.openapi.utils import get_openapi
+from fastapi.responses import JSONResponse
 from fastapi.routing import APIRoute
 from fastapi.staticfiles import StaticFiles
 from fastapi_mcp import FastApiMCP
@@ -12,14 +15,16 @@
 
 from alembic import command
 from apps.api import api_router
-from common.utils.embedding_threads import fill_empty_table_and_ds_embeddings
+from apps.swagger.i18n import PLACEHOLDER_PREFIX, tags_metadata
+from apps.swagger.i18n import get_translation, DEFAULT_LANG
 from apps.system.crud.aimodel_manage import async_model_info
 from apps.system.crud.assistant import init_dynamic_cors
 from apps.system.middleware.auth import TokenMiddleware
 from common.core.config import settings
 from common.core.response_middleware import ResponseMiddleware, exception_handler
 from common.core.sqlbot_cache import init_sqlbot_cache
-from common.utils.embedding_threads import fill_empty_terminology_embeddings, fill_empty_data_training_embeddings
+from common.utils.embedding_threads import fill_empty_terminology_embeddings, fill_empty_data_training_embeddings, \
+    fill_empty_table_and_ds_embeddings
 from common.utils.utils import SQLBotLogUtil
 
 
@@ -65,9 +70,104 @@ def custom_generate_unique_id(route: APIRoute) -> str:
     title=settings.PROJECT_NAME,
     openapi_url=f"{settings.API_V1_STR}/openapi.json",
     generate_unique_id_function=custom_generate_unique_id,
-    lifespan=lifespan
+    lifespan=lifespan,
+    docs_url=None,
+    redoc_url=None
 )
 
+# cache docs for different text
+_openapi_cache: Dict[str, Dict[str, Any]] = {}
+
+# replace placeholder
+def replace_placeholders_in_schema(schema: Dict[str, Any], trans: Dict[str, str]) -> None:
+    """
+    search OpenAPI schema，replace PLACEHOLDER_xxx to text。
+    """
+    if isinstance(schema, dict):
+        for key, value in schema.items():
+            if isinstance(value, str) and value.startswith(PLACEHOLDER_PREFIX):
+                placeholder_key = value[len(PLACEHOLDER_PREFIX):]
+                schema[key] = trans.get(placeholder_key, value)
+            else:
+                replace_placeholders_in_schema(value, trans)
+    elif isinstance(schema, list):
+        for item in schema:
+            replace_placeholders_in_schema(item, trans)
+
+
+
+# OpenAPI build
+def get_language_from_request(request: Request) -> str:
+    # get param from query ?lang=zh
+    lang = request.query_params.get("lang")
+    if lang in ["en", "zh"]:
+        return lang
+    # get lang from Accept-Language Header
+    accept_lang = request.headers.get("accept-language", "")
+    if "zh" in accept_lang.lower():
+        return "zh"
+    return DEFAULT_LANG
+
+
+def generate_openapi_for_lang(lang: str) -> Dict[str, Any]:
+    if lang in _openapi_cache:
+        return _openapi_cache[lang]
+
+    # tags metadata
+    trans = get_translation(lang)
+    localized_tags = []
+    for tag in tags_metadata:
+        desc = tag["description"]
+        if desc.startswith(PLACEHOLDER_PREFIX):
+            key = desc[len(PLACEHOLDER_PREFIX):]
+            desc = trans.get(key, desc)
+        localized_tags.append({
+            "name": tag["name"],
+            "description": desc
+        })
+
+    # 1. create OpenAPI
+    openapi_schema = get_openapi(
+        title="SQLBot API Document" if lang == "en" else "SQLBot API 文档",
+        version="1.0.0",
+        routes=app.routes,
+        tags=localized_tags
+    )
+
+    # openapi version
+    openapi_schema.setdefault("openapi", "3.1.0")
+
+    # 2. get trans for lang
+    trans = get_translation(lang)
+
+    # 3. replace placeholder
+    replace_placeholders_in_schema(openapi_schema, trans)
+
+    # 4. cache
+    _openapi_cache[lang] = openapi_schema
+    return openapi_schema
+
+
+
+# custom /openapi.json and /docs
+@app.get("/openapi.json", include_in_schema=False)
+async def custom_openapi(request: Request):
+    lang = get_language_from_request(request)
+    schema = generate_openapi_for_lang(lang)
+    return JSONResponse(schema)
+
+
+@app.get("/docs", include_in_schema=False)
+async def custom_swagger_ui(request: Request):
+    lang = get_language_from_request(request)
+    from fastapi.openapi.docs import get_swagger_ui_html
+    return get_swagger_ui_html(
+        openapi_url=f"/openapi.json?lang={lang}",
+        title="SQLBot API Docs",
+        swagger_favicon_url="https://fastapi.tiangolo.com/img/favicon.png",
+    )
+
+
 mcp_app = FastAPI()
 # mcp server, images path
 images_path = settings.MCP_IMAGE_PATH