Skip to content

Analysis flow

Jump to bottom
tangkikodo edited this page Jan 29, 2026 · 1 revision

Analysis.py 执行流程图

scan() 方法流程

flowchart TD
    A[开始 scan] --> B[重置状态<br/>expose_set, collect_set, metadata]
    B --> C[获取 root_class 的核心类型<br/>get_core_types]
    C --> D[更新前向引用<br/>update_forward_refs]
    D --> E[遍历所有核心类型<br/>调用 _walker]

    E --> F[返回 metadata]

    style A fill:#e1f5ff
    style F fill:#d4edda
Loading

_walker() 方法详细流程

flowchart TD
    A[开始 _walker] --> B[获取类的完全限定名<br/>kls_name]
    B --> C{检查缓存<br/>metadata.get kls_name}

    C -->|已存在| D{检查是否需要遍历<br/>should_traverse or has_config}
    D -->|需要| E[填充祖先信息<br/>_populate_ancestors]
    D -->|不需要| F[直接返回]
    E --> F

    C -->|不存在| G[准备 ER 图<br/>er_pre_generator.prepare]
    G --> H[生成 expose 配置<br/>pre_generate_expose_config]
    H --> I[生成 collector 配置<br/>pre_generate_collector_config]

    I --> J[提取字段信息<br/>_get_all_fields_and_object_fields]
    J --> K[提取方法信息<br/>resolve_fields, post_fields]
    K --> L[验证字段存在性]

    L --> M[获取 expose_dict<br/>和 collect_dict]
    M --> N[验证 expose 配置]

    N --> O[扫描 resolve 方法参数<br/>_scan_resolve_method]
    O --> P[扫描 post 方法参数<br/>_scan_post_method]
    P --> Q[扫描 post_default_handler<br/>_scan_post_default_handler]

    Q --> R{检查是否有 context<br/>resolve or post}
    R --> S{检查 collector 配置}

    S --> T[构建并存储元数据字典<br/>KlsMetaType]
    T --> U[metadata kls_name = info]

    U --> V{遍历无 resolver 的字段<br/>object_fields_without_resolver}
    V --> W[递归调用 _walker<br/>处理嵌套对象]

    W --> X{遍历有 resolver 的字段<br/>fields_with_resolver}
    X --> Y[递归调用 _walker<br/>处理嵌套对象]

    Y --> Z{决定是否需要遍历<br/>should_traverse or has_config}
    Z -->|需要| AA[填充祖先信息<br/>_populate_ancestors]
    Z -->|不需要| AB[结束]
    AA --> AB[结束]

    style A fill:#e1f5ff
    style AB fill:#d4edda
    style C fill:#fff3cd
    style F fill:#f8d7da
Loading

关键数据结构

元数据字典 (KlsMetaType)

{
    'resolve': ['resolve_field1', 'resolve_field2'],  # resolve 方法名列表
    'resolve_params': {  # resolve 方法的参数信息
        'resolve_field1': {
            'trim_field': 'field1',
            'context': bool,
            'parent': bool,
            'ancestor_context': bool,
            'dataloaders': [...]
        }
    },
    'post': ['post_field1', 'post_field2'],  # post 方法名列表
    'post_params': {  # post 方法的参数信息
        'post_field1': {
            'trim_field': 'field1',
            'context': bool,
            'parent': bool,
            'ancestor_context': bool,
            'dataloaders': [...],
            'collectors': [...]
        }
    },
    'post_default_handler_params': {...},  # post_default_handler 参数
    'raw_object_fields': ['obj_field1', 'obj_field2'],  # 没有 resolver 的字段
    'object_fields': [],  # 对象字段列表（待填充）
    'expose_dict': {'field': 'alias'},  # expose 配置
    'collect_dict': {'field': 'collector_alias'},  # collect 配置
    'kls': ModelClass,  # Pydantic 模型类
    'has_context': bool,  # 是否需要 context
    'should_traverse': bool  # 是否需要遍历
}

执行顺序示例

场景：解析一个有嵌套关系的模型

class User(BaseModel):
    id: int
    name: str

class Comment(BaseModel):
    id: int
    content: str
    user_id: int
    user: User = None

    def resolve_user(self, loader=Loader(user_loader)):
        return loader.load(self.user_id)

class Post(BaseModel):
    id: int
    title: str
    comments: List[Comment] = []

    def resolve_comments(self, loader=Loader(comments_loader)):
        return loader.load(self.id)

执行流程

sequenceDiagram
    participant S as scan()
    participant W as _walker()
    participant M as metadata

    S->>S: 重置状态
    S->>S: 获取核心类型 [Post]

    S->>W: _walker(Post, [])

    Note over W: 第1次遍历 - Post
    W->>W: 检查缓存 - 不存在
    W->>W: 准备配置 (expose/collect)
    W->>W: 提取字段: id, title, comments
    W->>W: 提取方法: resolve_comments
    W->>W: 扫描 resolve_comments 参数
    W->>M: 存储 Post 元数据
    W->>W: 遍历嵌套对象: List[Comment]

    W->>W: _walker(Comment, [(comments, Post)])

    Note over W: 第2次遍历 - Comment
    W->>W: 检查缓存 - 不存在
    W->>W: 准备配置
    W->>W: 提取字段: id, content, user_id
    W->>W: 提取方法: resolve_user
    W->>W: 扫描 resolve_user 参数
    W->>M: 存储 Comment 元数据
    W->>W: 遍历嵌套对象: User

    W->>W: _walker(User, [(user, Comment)])

    Note over W: 第3次遍历 - User
    W->>W: 检查缓存 - 不存在
    W->>W: 准备配置
    W->>W: 提取字段: id, name
    W->>W: 无 resolve/post 方法
    W->>M: 存储 User 元数据
    W->>W: 无嵌套对象 - 结束

    W-->>M: 完成
    S->>S: 返回 metadata
Loading

关键设计点

1. 广度优先遍历

  • 先处理当前类的所有信息
  • 然后递归处理嵌套对象

2. 两次遍历字段 

# 先遍历没有 resolver 的字段
for field, shelled_types in (obj for obj in object_fields
                              if obj[0] in object_fields_without_resolver):
    ...

# 再遍历有 resolver 的字段
for field, shelled_types in (obj for obj in object_fields
                              if obj[0] in fields_with_resolver):
    ...

原因：确保字段的解析顺序，有 resolver 的字段会覆盖没有 resolver 的字段

3. 缓存机制 

cached_metadata = self.metadata.get(kls_name)
if cached_metadata:
    # 已处理过，直接返回或只更新祖先信息
    ...

好处：

  • 避免重复处理同一个类
  • 支持循环引用
  • 提升性能

4. should_traverse 标记 

should_traverse = info['should_traverse'] or self._has_config(info)

含义：标记这个类是否需要在解析时被"遍历"（调用其 resolve/post 方法）

何时为 True：

  • 类有 resolve/post 方法
  • 类有 expose/collect 配置

数据流向

graph LR
    A[root_class] --> B[scan]
    B --> C[_walker]
    C --> D[metadata]
    D --> E[convert_metadata_key_as_kls]
    C --> F[递归调用 _walker]
    F --> D
    E --> G[Resolver 使用]
Loading

性能优化点

  1. 缓存：已处理的类不再重复扫描
  2. 惰性计算：只在需要时才扫描方法参数
  3. 批量处理：一次处理所有 resolve/post 方法
  4. 类型解析优化：通过 get_core_types 处理 Union 类型

Clone this wiki locally