style: 一些复杂而综合的细节修正
This commit is contained in:
@@ -38,17 +38,21 @@ import inspect
|
||||
import logging
|
||||
from typing import Any, Dict, Optional, Set
|
||||
|
||||
from docstring_parser import DocstringStyle, ParseError, parse
|
||||
from docstring_parser import Docstring, DocstringStyle, ParseError, parse
|
||||
from fastapi import FastAPI
|
||||
from fastapi.openapi.utils import get_openapi
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _parse_docstring(
|
||||
func: Any, style: DocstringStyle = DocstringStyle.AUTO
|
||||
) -> Optional[Any]:
|
||||
"""解析函数的 Google Style docstring，失败时返回 None。"""
|
||||
def _parse_docstring(func: Any, style: DocstringStyle = DocstringStyle.AUTO) -> Optional[Docstring]: # noqa ANN401 有意为之
|
||||
"""
|
||||
解析函数的 Google Style docstring，失败时返回 None。
|
||||
|
||||
Returns:
|
||||
成功解析时返回 docstring_parser.Docstring。
|
||||
失败时返回 None。
|
||||
"""
|
||||
doc = inspect.getdoc(func)
|
||||
if not doc:
|
||||
return None
|
||||
@@ -67,8 +71,13 @@ def _parse_docstring(
|
||||
return None
|
||||
|
||||
|
||||
def _build_param_lookup(parsed_doc: Any) -> Dict[str, str]:
|
||||
"""从解析后的 docstring 构建 {参数名: 描述} 映射。"""
|
||||
def _build_param_lookup(parsed_doc: Docstring) -> Dict[str, str]:
|
||||
"""
|
||||
从解析后的 docstring 构建 {参数名: 描述} 映射。
|
||||
|
||||
Returns:
|
||||
{参数名: 描述}
|
||||
"""
|
||||
lookup: Dict[str, str] = {}
|
||||
for param in getattr(parsed_doc, "params", []) or []:
|
||||
name = getattr(param, "arg_name", None)
|
||||
@@ -79,8 +88,13 @@ def _build_param_lookup(parsed_doc: Any) -> Dict[str, str]:
|
||||
return lookup
|
||||
|
||||
|
||||
def _build_returns_description(parsed_doc: Any) -> Optional[str]:
|
||||
"""从解析后的 docstring 构建返回值描述。"""
|
||||
def _build_returns_description(parsed_doc: Docstring) -> Optional[str]:
|
||||
"""
|
||||
从解析后的 docstring 构建返回值描述。
|
||||
|
||||
Returns:
|
||||
可能的字符串为返回值描述。
|
||||
"""
|
||||
ret = getattr(parsed_doc, "returns", None)
|
||||
if not ret:
|
||||
return None
|
||||
@@ -91,8 +105,13 @@ def _build_returns_description(parsed_doc: Any) -> Optional[str]:
|
||||
return desc or None
|
||||
|
||||
|
||||
def _build_raises_description(parsed_doc: Any) -> Optional[str]:
|
||||
"""从解析后的 docstring 构建异常描述（Markdown 列表）。"""
|
||||
def _build_raises_description(parsed_doc: Docstring) -> Optional[str]:
|
||||
"""
|
||||
从解析后的 docstring 构建异常描述（Markdown 列表）。
|
||||
|
||||
Returns:
|
||||
可能的字符串为异常描述。
|
||||
"""
|
||||
raises = getattr(parsed_doc, "raises", None)
|
||||
if not raises:
|
||||
return None
|
||||
@@ -110,7 +129,12 @@ def _build_raises_description(parsed_doc: Any) -> Optional[str]:
|
||||
|
||||
|
||||
def _dedent_description(text: str) -> str:
|
||||
"""清理 docstring 描述中的多余缩进，保留段落结构。"""
|
||||
"""
|
||||
清理 docstring 描述中的多余缩进，保留段落结构。
|
||||
|
||||
Returns:
|
||||
docstring 字符串
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
lines = text.splitlines()
|
||||
@@ -123,11 +147,12 @@ def _dedent_description(text: str) -> str:
|
||||
return "\n".join(cleaned).strip()
|
||||
|
||||
|
||||
def _get_clean_operation_description(parsed_doc: Any) -> str:
|
||||
def _get_clean_operation_description(parsed_doc: Docstring) -> str:
|
||||
"""
|
||||
生成干净的 operation description。
|
||||
|
||||
策略：保留 summary + long_description，去掉 Args/Returns/Raises 等机器块。
|
||||
Returns:
|
||||
策略：保留 summary + long_description，去掉 Args/Returns/Raises 等机器块。
|
||||
"""
|
||||
parts: list[str] = []
|
||||
short = getattr(parsed_doc, "short_description", None)
|
||||
@@ -142,10 +167,10 @@ def _get_clean_operation_description(parsed_doc: Any) -> str:
|
||||
|
||||
|
||||
def _inject_schema_properties(
|
||||
schema: Dict[str, Any],
|
||||
param_lookup: Dict[str, str],
|
||||
visited_refs: Optional[Set[str]] = None,
|
||||
openapi_schema: Optional[Dict[str, Any]] = None,
|
||||
schema: Dict[str, Any],
|
||||
param_lookup: Dict[str, str],
|
||||
visited_refs: Optional[Set[str]] = None,
|
||||
openapi_schema: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
"""
|
||||
递归注入 schema properties 的描述。
|
||||
@@ -174,7 +199,7 @@ def _inject_schema_properties(
|
||||
properties = schema.get("properties")
|
||||
if isinstance(properties, dict):
|
||||
for prop_name, prop_schema in properties.items():
|
||||
if prop_name in param_lookup:
|
||||
if prop_name in param_lookup: # noqa SIM102
|
||||
# 只有当 docstring 描述非空时才写入
|
||||
if param_lookup[prop_name]:
|
||||
prop_schema["description"] = param_lookup[prop_name]
|
||||
@@ -198,11 +223,11 @@ def _inject_schema_properties(
|
||||
|
||||
|
||||
def enrich_openapi_from_docstrings(
|
||||
app: FastAPI,
|
||||
*,
|
||||
prefer_docstring_over_field: bool = True,
|
||||
append_raises_to_description: bool = True,
|
||||
docstring_style: DocstringStyle = DocstringStyle.AUTO,
|
||||
app: FastAPI,
|
||||
*,
|
||||
prefer_docstring_over_field: bool = True,
|
||||
append_raises_to_description: bool = True,
|
||||
docstring_style: DocstringStyle = DocstringStyle.AUTO,
|
||||
) -> None:
|
||||
"""
|
||||
为 FastAPI 应用启用 docstring 驱动的 OpenAPI 增强。
|
||||
@@ -220,8 +245,8 @@ def enrich_openapi_from_docstrings(
|
||||
解析风格。团队统一用 Google 时可显式传入 ``DocstringStyle.GOOGLE``。
|
||||
"""
|
||||
|
||||
# 保存原始的 openapi 函数（如果有）
|
||||
original_openapi = app.openapi
|
||||
# 可选择保存原始的 openapi 函数（如果有）
|
||||
# original_openapi = app.openapi
|
||||
|
||||
def custom_openapi() -> Dict[str, Any]:
|
||||
# 如果已经有缓存，直接返回
|
||||
@@ -284,9 +309,7 @@ def enrich_openapi_from_docstrings(
|
||||
for media_obj in content.values():
|
||||
schema = media_obj.get("schema", {})
|
||||
if schema:
|
||||
_inject_schema_properties(
|
||||
schema, param_lookup, openapi_schema=openapi_schema
|
||||
)
|
||||
_inject_schema_properties(schema, param_lookup, openapi_schema=openapi_schema)
|
||||
|
||||
# ---------- 4. 注入返回值描述 ----------
|
||||
if returns_desc:
|
||||
@@ -300,9 +323,7 @@ def enrich_openapi_from_docstrings(
|
||||
if schema and not schema.get("description"):
|
||||
schema["description"] = returns_desc
|
||||
# 递归注入 schema 内部字段
|
||||
_inject_schema_properties(
|
||||
schema, param_lookup, openapi_schema=openapi_schema
|
||||
)
|
||||
_inject_schema_properties(schema, param_lookup, openapi_schema=openapi_schema)
|
||||
|
||||
# ---------- 5. 异常描述追加 ----------
|
||||
if append_raises_to_description and raises_desc:
|
||||
@@ -315,61 +336,4 @@ def enrich_openapi_from_docstrings(
|
||||
app.openapi_schema = openapi_schema
|
||||
return app.openapi_schema
|
||||
|
||||
app.openapi = custom_openapi
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 完整使用示例
|
||||
# =============================================================================
|
||||
if __name__ == "__main__":
|
||||
import json
|
||||
from typing import Annotated, Optional
|
||||
|
||||
from fastapi import FastAPI, Path, Query
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
# ---------- 定义 DTO ----------
|
||||
class EditChatDto(BaseModel):
|
||||
old_message: str = Field(..., description="原始消息内容")
|
||||
new_message: str = Field(..., description="修改后的消息内容")
|
||||
change: str = Field(..., description="变更类型")
|
||||
|
||||
class ReturnDto(BaseModel):
|
||||
result: str = Field(..., description="操作结果")
|
||||
content: str = Field(..., description="最新聊天记录内容")
|
||||
|
||||
# ---------- 创建应用 ----------
|
||||
app = FastAPI(title="Chatroom API", version="1.0.0")
|
||||
|
||||
# 启用 docstring 增强（必须在注册路由之前或之后都可以，但要在生成 schema 之前）
|
||||
enrich_openapi_from_docstrings(app)
|
||||
|
||||
# ---------- 注册路由 ----------
|
||||
@app.post("/api/chatroom/{id_}/chat/edit/", response_model=ReturnDto)
|
||||
async def edit_chatroom_chat(
|
||||
id_: Annotated[str, Path(description="聊天室 ID")],
|
||||
body: EditChatDto,
|
||||
force: Annotated[Optional[bool], Query(description="是否强制覆盖")] = None,
|
||||
) -> ReturnDto:
|
||||
"""编辑聊天室消息。
|
||||
|
||||
此端点不负责调用 AI 生成输出，而是用于修改一条已经保存在聊天记录中的消息。
|
||||
前端调用后应使用返回的 content 刷新当前聊天室界面。
|
||||
|
||||
Args:
|
||||
id_: 聊天室唯一标识符，UUID 格式。
|
||||
body: 编辑请求体，包含旧消息、新消息和变更类型。
|
||||
force: 是否跳过冲突检测直接覆盖。
|
||||
|
||||
Returns:
|
||||
ReturnDto: 操作结果，其中 result 字段表示状态，content 字段为最新聊天记录。
|
||||
|
||||
Raises:
|
||||
HTTPException: 404 表明未找到聊天室。
|
||||
HTTPException: 400 表明聊天记录匹配失败，未更新。
|
||||
"""
|
||||
return ReturnDto(result="ok", content="new content")
|
||||
|
||||
# ---------- 输出生成的 OpenAPI schema ----------
|
||||
schema = app.openapi()
|
||||
print(json.dumps(schema, indent=2, ensure_ascii=False))
|
||||
app.openapi = custom_openapi # type: ignore[method-assign]
|
||||
|
||||
Reference in New Issue
Block a user