Python框架开发规范：类型提示、单元测试与文档生成最佳实践

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "hyperflow-core"
version = "1.0.0"
description = "A strictly-typed high performance workflow framework"
readme = "README.md"
requires-python = ">=3.9"

[tool.hatch.build.targets.wheel]
packages = ["src/hyperflow"]
include = [
    "src/hyperflow/py.typed",
]

from typing import Protocol, runtime_checkable, TypeVar, Any
from collections.abc import Mapping

T_co = TypeVar("T_co", covariant=True)

@runtime_checkable
class Serializer(Protocol[T_co]):
    """序列化器结构化子类型协议。
    
    下游组件无需显式继承此类，仅需实现对应签名即可通过静态校验。
    """
    def serialize(self, data: Mapping[str, Any]) -> T_co:
        ...

    def validate_schema(self, payload: bytes) -> bool:
        ...

import pytest
from typing import Generator
from hyperflow.engine import PipelineContext, reset_global_registry

@pytest.fixture(scope="function")
def isolated_context() -> Generator[PipelineContext, None, None]:
    """提供完全隔离的上下文实例，并在测试结束后执行清理。"""
    ctx = PipelineContext(name="test_pipeline", debug=True)
    yield ctx
    # Teardown 阶段：重置注册表，清理挂载的临时句柄
    ctx.teardown()
    reset_global_registry()

import pytest
from hyperflow.exceptions import ConfigurationError, ValidationError
from hyperflow.pipeline import Pipeline

@pytest.mark.parametrize(
    "invalid_input, expected_err, match_pattern",
    [
        ("", ConfigurationError, "Pipeline name cannot be empty"),
        (" ", ConfigurationError, "Pipeline name cannot be empty"),
        ({"invalid": "schema"}, ValidationError, r"Missing required field 'steps'"),
    ],
    ids=["empty_name", "whitespace_name", "missing_steps"]
)
def test_pipeline_invalid_initialization(
    invalid_input: str | dict,
    expected_err: type[Exception],
    match_pattern: str
) -> None:
    with pytest.raises(expected_err, match=match_pattern):
        Pipeline(invalid_input)  # type: ignore[arg-type]

import nox

# 定义需要保证严格兼容的解释器版本列表
PYTHON_VERSIONS = ["3.9", "3.10", "3.11", "3.12", "3.13"]

@nox.session(python=PYTHON_VERSIONS)
def tests(session: nox.Session) -> None:
    """在各个纯净虚拟环境中安装依赖并运行测试。"""
    session.install(".[test]")
    session.run(
        "pytest",
        "--cov=hyperflow",
        "--cov-branch",
        "--cov-report=term-missing",
        "--cov-fail-under=95",
        "tests"
    )

@nox.session(python="3.11")
def typecheck(session: nox.Session) -> None:
    """运行静态类型门禁检查。"""
    session.install(".[dev]")
    session.run("mypy", "--strict", "src/hyperflow")

def execute_task(
    task_id: str,
    payload: dict[str, Any],
    *,
    timeout: float = 30.0,
    retry_count: int = 3
) -> ExecutionResult:
    """调度并执行指定的流水线子任务。

    使用内置的工作池调度执行，支持重试策略与超时截断。

    Args:
        task_id: 唯一任务标识符，不可为空字符串。
        payload: 传递给任务执行上下文的字典载荷。
        timeout: 单次执行的最长超时时间（秒），必须大于 0。
        retry_count: 遇到瞬时网络异常时的最大重试次数。

    Returns:
        包含执行状态码、输出结果与耗时指标的 ExecutionResult 对象。

    Raises:
        ValueError: 当 task_id 为空或 timeout 小于等于 0 时抛出。
        TaskTimeoutError: 当重试耗尽且任务仍超时时抛出。

    Examples:
        >>> from hyperflow.pipeline import execute_task
        >>> res = execute_task("task_01", {"action": "sync"}, timeout=5.0)
        >>> res.is_success
        True
    """
    if not task_id or timeout <= 0:
        raise ValueError("Invalid task_id or timeout")
    # 具体执行逻辑...
    return ExecutionResult(task_id=task_id, is_success=True)

site_name: Hyperflow 官方开发指南
theme:
  name: material
  palette:
    - scheme: default
      primary: black
      accent: deep orange
  features:
    - content.code.copy
    - search.suggest

plugins:
  - search
  - mkdocstrings:
      default_handler: python
      handlers:
        python:
          paths: [src]
          options:
            show_root_heading: true
            show_source: true
            members_order: source
            merge_init_into_class: true
            docstring_style: google
            show_signature_annotations: true

pytest --doctest-modules src/hyperflow