# 编码规范 本文档定义新 SaaS 项目的 Python 编码规范。 --- ## 1. 基础规范 遵循 [PEP 8](https://peps.python.org/pep-0008/),以下是重点和补充。 ### 1.1 命名规范 **模块/包**:小写 + 下划线 ```python # ✅ 正确 from packages.domain import entities from packages.adapters.in_memory import project_repository # ❌ 错误 from packages.Domain import Entities from packages.adapters.InMemory import ProjectRepository ``` **类**:PascalCase ```python # ✅ 正确 class Project: pass class InMemoryProjectRepository: pass # ❌ 错误 class project: pass class in_memory_project_repository: pass ``` **函数/变量**:小写 + 下划线 ```python # ✅ 正确 def create_project(workspace_id: str, name: str) -> Project: pass user_count = 10 # ❌ 错误 def CreateProject(workspace_id: str, name: str) -> Project: pass UserCount = 10 ``` **常量**:大写 + 下划线 ```python # ✅ 正确 MAX_PROJECT_NAME_LENGTH = 100 DEFAULT_PAGE_SIZE = 20 # ❌ 错误 maxProjectNameLength = 100 default_page_size = 20 ``` **私有属性/方法**:前缀 `_` ```python class Project: def __init__(self): self._internal_state = {} def _validate(self): pass ``` --- ## 2. Type Hints **强制使用** type hints,提升代码可读性和 IDE 支持。 ```python # ✅ 正确 def create_project(workspace_id: str, name: str, description: str = "") -> Project: pass def list_projects(workspace_id: str) -> list[Project]: pass def get_project(project_id: str) -> Project | None: pass # ❌ 错误 def create_project(workspace_id, name, description=""): pass ``` **复杂类型**: ```python from typing import Protocol, Any # Dict/List def update_metadata(metadata: dict[str, Any]) -> None: pass # Optional (Python 3.10+ 用 | None) def get_user(user_id: str) -> User | None: pass # Protocol class Repository(Protocol): def get(self, id: str) -> Entity | None: pass ``` --- ## 3. Dataclass **优先使用** `dataclass` 定义实体和值对象。 ```python from dataclasses import dataclass, field from datetime import datetime, timezone # ✅ 正确 @dataclass(slots=True) class Project: id: str workspace_id: str name: str description: str = "" created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) # ❌ 错误(不用 dataclass) class Project: def __init__(self, id: str, workspace_id: str, name: str, description: str = ""): self.id = id self.workspace_id = workspace_id self.name = name self.description = description ``` **为什么用 `slots=True`?** - 节省内存 - 防止意外添加属性 - 提升性能 --- ## 4. 注释与文档 ### 4.1 模块/类/函数注释 **使用中文注释**。 ```python def create_project(workspace_id: str, name: str, description: str = "") -> Project: """ 创建项目。 Args: workspace_id: 工作空间 ID name: 项目名称(不能为空) description: 项目描述(可选) Returns: 创建的项目实体 Raises: ValueError: 项目名称为空时 """ pass ``` ### 4.2 复杂逻辑注释 ```python # 正确做法:复杂逻辑加注释 def calculate_priority(job: IngestJob) -> int: # 优先级规则: # 1. FAILED 状态最高(需要重试) # 2. PENDING 状态次高(等待处理) # 3. PROCESSING 状态最低(正在处理) if job.status == IngestJobStatus.FAILED: return 100 elif job.status == IngestJobStatus.PENDING: return 50 else: return 10 ``` --- ## 5. 异常处理 ### 5.1 使用具体异常 ```python # ✅ 正确 def get_project(project_id: str) -> Project: if not project_id: raise ValueError("项目 ID 不能为空") project = repository.get(project_id) if project is None: raise KeyError(f"项目 {project_id} 不存在") return project # ❌ 错误 def get_project(project_id: str) -> Project: if not project_id: raise Exception("错误") # 太宽泛 ``` ### 5.2 自定义异常 ```python class ProjectNotFoundError(Exception): """项目不存在异常。""" pass class ProjectNameTooLongError(ValueError): """项目名称过长异常。""" pass ``` --- ## 6. Clean Architecture 约束 ### 6.1 依赖方向 ``` Apps (api/worker/web) ↓ Application (use cases) ↓ Ports (interfaces) ← Adapters (implementations) ↓ Domain (entities/rules) ``` **Domain 层**: - ❌ 不能依赖任何外层 - ❌ 不能依赖 SQLAlchemy、FastAPI、Celery - ✅ 只能依赖 Python 标准库 ```python # ✅ 正确(Domain 层) from dataclasses import dataclass from datetime import datetime from uuid import uuid4 @dataclass(slots=True) class Project: id: str name: str # ❌ 错误(Domain 层) from sqlalchemy import Column, String # ❌ 不能依赖 SQLAlchemy from fastapi import HTTPException # ❌ 不能依赖 FastAPI @dataclass(slots=True) class Project: id: str name: str ``` **Application 层**: - ✅ 可以依赖 Domain + Ports - ❌ 不能依赖 Adapters **Adapters 层**: - ✅ 可以依赖 Domain + Ports - ✅ 可以使用外部库(SQLAlchemy、Redis 等) --- ## 7. 测试 ### 7.1 测试文件命名 ``` tests/ ├── integration/ │ ├── test_projects.py │ ├── test_ingest_pipeline.py │ └── test_classification_pipeline.py └── unit/ ├── test_project_entity.py └── test_asset_validation.py ``` ### 7.2 测试函数命名 ```python # ✅ 正确 def test_create_project_with_valid_name(): pass def test_create_project_with_empty_name_should_fail(): pass def test_list_projects_by_workspace(): pass # ❌ 错误 def test1(): pass def test_project(): pass ``` ### 7.3 测试结构(AAA 模式) ```python def test_create_project(): # Arrange(准备) workspace_id = "ws-1" name = "测试项目" repository = InMemoryProjectRepository() use_case = CreateProjectUseCase(repository) # Act(执行) project = use_case.execute( CreateProjectCommand(workspace_id=workspace_id, name=name) ) # Assert(断言) assert project.name == name assert project.workspace_id == workspace_id ``` --- ## 8. 代码格式化 ### 8.1 行长度 - 最大 120 字符 - 优先 88 字符(Black 默认) ### 8.2 导入顺序 ```python # 1. 标准库 import os from datetime import datetime from typing import Protocol # 2. 第三方库 from fastapi import FastAPI from sqlalchemy import Column # 3. 本地模块 from packages.domain import Project from packages.application import CreateProjectUseCase ``` ### 8.3 空行 ```python # 类之间:2 行 class User: pass class Workspace: pass # 函数之间:1 行 def create_user(): pass def list_users(): pass ``` --- ## 9. 安全规范 ### 9.1 禁止硬编码敏感信息 ```python # ❌ 错误 DATABASE_URL = "postgresql://admin:password123@localhost/db" API_KEY = "sk-1234567890abcdef" # ✅ 正确 import os DATABASE_URL = os.getenv("DATABASE_URL") API_KEY = os.getenv("API_KEY") ``` ### 9.2 输入验证 ```python # ✅ 正确 def create_project(name: str) -> Project: clean_name = name.strip() if not clean_name: raise ValueError("项目名称不能为空") if len(clean_name) > 100: raise ValueError("项目名称不能超过 100 字符") return Project(id=uuid4().hex, name=clean_name) ``` --- ## 10. 性能规范 ### 10.1 避免 N+1 查询 ```python # ❌ 错误 projects = repository.list_by_workspace("ws-1") for project in projects: assets = asset_repository.list_by_project(project.id) # N+1 # ✅ 正确 projects = repository.list_by_workspace("ws-1") project_ids = [p.id for p in projects] assets = asset_repository.list_by_projects(project_ids) # 一次查询 ``` ### 10.2 使用生成器 ```python # ✅ 正确(大数据集) def list_all_assets() -> Generator[Asset, None, None]: for asset in repository.stream(): yield asset # ❌ 错误(加载全部到内存) def list_all_assets() -> list[Asset]: return repository.list_all() # 可能 OOM ``` --- **最后更新**: 2026-06-15 **版本**: v1.0