writing-python-code

编写 Python 代码

前提条件

此技能扩展了 myai 的 engineering-principles。请先加载该技能。假定已通过 myai 引导程序加载 using-my-skills 和 engineering-principles。

关于通用工程理念（成功之路、快速失败、将错误处理视为控制流、测试理念、架构分离），请参阅 myai 的 engineering-principles。此技能仅涵盖 Python 专属的工具、类型系统配置和模式。

通过 uv 运行项目本地工具，而不要使用系统安装的二进制文件：uv run python、uv run pytest、uv run ruff、uv run basedpyright、uv run poe、uv run pre-commit。常规验证组合是 uv run poe lint_full 和 uv run poe test。

---

类型系统

basedpyright 配置

[tool.pyright]
pythonVersion = "3.14"
typeCheckingMode = "strict"
reportAny = "error"
reportImportCycles = "error"
reportImplicitStringConcatenation = "none"
reportUnusedCallResult = "none"
reportUnnecessaryIsInstance = "none"

适用于中大型项目的其他严格选项：

reportExplicitAny = "error"
reportUnnecessaryTypeIgnoreComment = "error"
reportMissingModuleSource = "error"
reportPrivateUsage = "error"
reportOptionalMemberAccess = "error"
reportOptionalCall = "error"
reportAttributeAccessIssue = "error"

禁用模式

禁用项	替代方案
Any	完全禁用——定义实际类型
object（不受限制）	仅限边界位置使用（见下文）
typing.cast()	isinstance、TypeIs、模式匹配
没有说明原因的 # type: ignore	# type: ignore[specific-code] # rationale: <reason>
业务逻辑中的原始 dict	msgspec.Struct、dataclass 或 TypedDict
隐式返回类型	为每个函数显式添加注解

仅允许在以下位置使用 object： TypeIs/TypeGuard 参数、*args: object、Coroutine[object, None, T]、PySide6 Signal(object)。其他所有位置都应使用 Protocol、TypeVar、联合类型或类型化结构。

类型模式

外部数据 → msgspec.Struct：

import msgspec

class UserConfig(msgspec.Struct):
    name: str
    port: int
    debug: bool = False

# JSON → typed object (validates at decode time)
config = msgspec.json.decode(raw_bytes, type=UserConfig)

# Dict/YAML → typed object
config = msgspec.convert(raw_dict, type=UserConfig)

当需要与字典兼容时（例如 **unpacking、要求使用字典的 API），仍可使用 TypedDict。

FastAPI HTTP 边界 → 在传输边界使用 pydantic DTO。立即将其转换为不依赖框架的类型化结构，不要让框架模型成为领域模型。

领域对象 → dataclass：

@dataclass(frozen=True, slots=True)
class Profile:
    name: str
    version: str
    active: bool = True

鸭子类型 → Protocol：

class Renderable(Protocol):
    def render(self) -> str: ...

常量 → Final：

import platformdirs

MAX_RETRIES: Final = 3
CONFIG_PATH: Final[Path] = Path(platformdirs.user_config_dir("app"))

处理库边界处的 Any

0. 类型化反序列化（用于外部数据）：

msgspec.json.decode(data, type=MyStruct) 可消除 JSON/API 响应中的 Any——返回类型是 MyStruct，而不是 Any。当边界是数据反序列化时，应先使用此方法，再考虑封装器。

1. 类型化封装器（库 API 的首选方案）：

class WhisperModelWrapper:
    def __init__(self, model_size: str, device: str = "auto") -> None:
        from faster_whisper import WhisperModel as _WhisperModel
        self._model = _WhisperModel(model_size, device=device)

    def transcribe(self, audio: np.ndarray, language: str | None = None) -> TranscriptionResult:
        segments_gen, info = self._model.transcribe(audio, language=language)
        return TranscriptionResult(
            text="".join(s.text for s in segments_gen),
            language=str(info.language),
        )

通过 ruff 强制使用封装器：

[tool.ruff.lint.flake8-tidy-imports.banned-api]
"faster_whisper" = { msg = "Use src/wrappers/whisper_wrapper instead" }

2. 类型存根，位于 src/stubs/：

# src/stubs/some_library.pyi
def some_function(arg: str) -> list[int]: ...

配置：在 basedpyright 配置中设置 stubPath = "src/stubs"。

3. 内联类型收窄，用于一次性场景：

raw_value = untyped_lib.get_value()  # returns Any
assert isinstance(raw_value, str)    # narrows to str

TypeIs 守卫

注意： 对于经 msgspec 解码的数据，无需使用 TypeIs 守卫（数据已完全类型化）。应使用它们来收窄类型未知的内存对象。

from typing import TypeIs, TypedDict, Required

class ValidResponse(TypedDict):
    status: Required[str]
    data: Required[dict[str, str | int | bool | list[str]]]

# Simplified for brevity — production code should use msgspec.convert() for full validation
def is_valid_response(obj: object) -> TypeIs[ValidResponse]:
    return (
        isinstance(obj, dict)
        and isinstance(obj.get("status"), str)
        and isinstance(obj.get("data"), dict)
    )

def process(response: object) -> Result[str, str]:
    if not is_valid_response(response):
        return Err("Invalid response format")
    return Ok(response["data"]["key"])  # type-safe access

使用模式匹配确保类型安全

@dataclass
class Success:
    value: str

@dataclass
class Failure:
    error: str
    code: int

type Outcome = Success | Failure

def handle(outcome: Outcome) -> str:
    match outcome:
        case Success(value=v):
            return f"OK: {v}"
        case Failure(error=e, code=c):
            return f"Error {c}: {e}"

# type: ignore 策略

每个 # type: ignore 都必须包含具体错误代码和原因说明：

# BAD
result = some_call()  # type: ignore

# GOOD
result = some_call()  # type: ignore[no-any-return]  # rationale: lib returns Any, validated below
assert isinstance(result, ExpectedType)

TYPE_CHECKING 守卫

对于会导致循环依赖或仅用于注解的导入：

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from src.managers.audio_manager import AudioManager

class TranscriptionService:
    def __init__(self, audio: "AudioManager") -> None: ...

注意： TYPE_CHECKING 用于同一层中的前向引用。如果两个模块相互导入，这就是循环依赖——应修正架构（提取一个 Protocol，将类型移至公共模块），而不是用 TYPE_CHECKING 隐藏循环依赖。

---

错误处理

关于将错误处理视为控制流的一般原则，请参阅 myai 的 engineering-principles。 Python 专属建议：对于预期会发生的失败，使用 rusty-results 中的 Result[T, E]。异常仅用于编程错误（bug）。

决策表

情况	使用
文件未找到、网络错误、输入无效	Result[T, E]
用户提供了错误数据	Result[T, E]
第三方库引发错误	在边界处捕获 → Result[T, E]
违反不变量（本不应发生）	raise 异常
程序状态无效（bug）	raise 异常

模式

import msgspec
from rusty_results import Result, Ok, Err

class Config(msgspec.Struct):
    name: str

def load_config(path: Path) -> Result[Config, str]:
    if not path.exists():
        return Err(f"Config not found: {path}")
    try:
        return Ok(msgspec.json.decode(path.read_bytes(), type=Config))
    except (msgspec.DecodeError, OSError) as e:
        return Err(f"Failed to load: {e}")

规则

1. 尽早返回——优先处理错误，让成功路径保持线性：

   result = load_data(path)
   if result.is_err:
       return Err(f"Cannot proceed: {result.unwrap_err()}")
   data = result.unwrap()
   

2. 三个错误边界：

   • 库：捕获第三方异常 → Result
   • 组件：每个子系统向调用方返回 Result
   • 全局：UI/CLI 顶层捕获所有异常，并向用户显示消息

3. 绝不吞掉错误——不要使用 except: pass，也不要忽略 Result。必须检查每个 Result[T, E]——至少记录错误，并在 GUI 中显示 toast/alert，或在 CLI 中打印错误。

4. 失败时清理——如果多步骤操作中途失败，应清理部分状态

5. 复杂领域使用自定义错误类型：

   @dataclass
   class ConfigError:
       path: Path
       reason: str
       line: int | None = None
   

6. 妥善处理收到的错误值：显示 UI 警告、记录日志、尽早返回，或附带上下文后继续传播

异步任务边界

通过 asyncio.ensure_future() 或 create_task() 启动的任何协程都是一个即发即忘边界。如果异常逸出，就不会有人处理它，UI 也会卡在中间状态（例如永远显示“处理中”）。

强制模式： 将整个协程主体包裹在 try/except Exception 中，作为安全网：

async def _do_work(self, path: Path) -> None:
    try:
        await self._do_work_inner(path)
    except Exception as exc:
        logger.exception("Unexpected error for %s", path)
        self._set_error_state(path, f"Unexpected error: {exc}")

内部方法处理预期错误（Result 检查、特定异常）。外部方法确保 UI 始终转换到终止状态。

CLI 错误边界

def main() -> int:
    result = run_app()
    if result.is_err:
        typer.echo(f"Error: {result.unwrap_err()}", err=True)
        return 1
    return 0

---

异步模式

适用场景

使用 async	不使用 async
文件 I/O	纯数据转换
网络请求	简单计算
子进程执行	内存操作
操作耗时 >100ms	快速查找
并行 I/O 操作	顺序纯逻辑

HTTP 请求

async def fetch_data(url: str) -> Result[bytes, str]:
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get(url, timeout=30.0)
            resp.raise_for_status()
            return Ok(resp.content)
    except httpx.HTTPError as e:
        return Err(f"HTTP error: {e}")

子进程执行

高级用法：超时、终止以及可靠地恢复输出，请参阅 python-reliable-subprocess-handling。 在 asyncio.timeout() 中调用 communicate() 时，如果发生取消，已读取的字节会被丢弃；而 kill() 不会终止子进程的子进程——这两种情况都在该文档中介绍。以下模式仅适用于简单、快速且无需超时的场景。

async def run_command(args: list[str]) -> Result[str, str]:
    try:
        process = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await process.communicate()
        if process.returncode != 0:
            return Err(f"Command failed: {stderr.decode()}")
        return Ok(stdout.decode())
    except FileNotFoundError:
        return Err(f"Command not found: {args[0]}")

并发操作

async def fetch_all(urls: list[str]) -> list[Result[bytes, str]]:
    return await asyncio.gather(*(fetch_data(url) for url in urls))

超时

async def fetch_with_timeout(url: str) -> Result[bytes, str]:
    try:
        async with asyncio.timeout(10):
            return await fetch_data(url)
    except TimeoutError:
        return Err(f"Timeout fetching {url}")

异步规则

1. 在异步上下文中绝不调用 subprocess.run()、time.sleep() 或同步 HTTP
2. 子进程调用中绝不使用 shell=True
3. 子进程始终使用 asyncio.create_subprocess_exec()
4. 资源管理（HTTP 客户端、文件句柄）始终使用 async with
5. 绝不在同一调用链中混用同步和异步
6. 对可能挂起的操作，始终处理 TimeoutError

---

代码风格

规则	值
行长度	120
缩进	4 个空格
引号	双引号（仅在避免转义时使用单引号）
导入顺序	标准库 → 第三方 → 本地（由 ruff 自动排序）

命名

元素	约定	示例
模块、函数、变量	snake_case	load_config、user_name
类、TypedDict	PascalCase	ProfileManager、UserConfig
常量	UPPER_SNAKE	MAX_RETRIES、DEFAULT_PORT
私有成员	_prefix	_internal_state

文档

公共 API 使用 Google 风格的文档字符串。注释说明为什么，而不是是什么。

---

跨平台

成熟的项目应同等支持 Windows、macOS 和 Linux。有意针对特定操作系统的项目除外；独立的小型脚本可以仅支持特定操作系统。

• 路径使用 pathlib。用户目录使用 platformdirs：user_config_dir、user_data_dir、user_cache_dir、user_log_dir、user_state_dir。
• 不使用仅支持 POSIX 的模块或系统调用：os.killpg、pwd、grp、fcntl、resource、signal.SIGKILL。
• 细微的操作系统差异 → 使用范围狭窄的 sys.platform / os.name 分支。复杂的操作系统专属子系统 → 使用 Protocol，并为每个操作系统实现一个版本，在启动时选定（参见 building-multi-ui-apps）。业务逻辑中绝不按平台分支。
• 通过 start_new_session + os.killpg 关闭子进程仅适用于 POSIX——只在 POSIX 目标环境中使用；跨平台应用需要为每个操作系统提供独立的关闭实现。

---

前置条件与验证

有关通用的快速失败原则，请参阅 myai 的 engineering-principles。 Python 特定做法：在子系统入口处使用 msgspec.Struct 验证数据形状，使用 isinstance / TypeIs 收窄类型，并使用 Result[T,E] 处理预期的验证失败。在继续执行业务逻辑之前，检查权限、外部依赖、配置有效性和输入范围。

---

架构

有关通用的架构分离原则，请参阅 myai 的 engineering-principles 和 architecting-changes。 Python 特定的布局：

Presentation (Qt GUI / CLI / API)
        |
        v
Domain (Managers, Models, Business Rules)
        |
        v
Utilities (Helpers, Wrappers, Common)

• 依赖只能向下流动
• UI 是一个插件——添加 CLI、GUI 或 API 不应改变业务逻辑
• 领域层绝不导入表示层
• 做出架构决策时，加载 architecting-python-changes（它扩展了 myai 的 architecting-changes）

---

安全性

• 子进程调用中不使用 shell=True
• 验证路径（考虑符号链接）：

  def is_safe_path(path: Path, base: Path) -> bool:
      try:
          resolved = path.resolve(strict=True)
          return resolved.is_relative_to(base.resolve(strict=True))
      except (OSError, ValueError):
          return False
  

• 不硬编码密钥——使用环境变量或 dotenv
• 在系统边界进行输入验证
• 绝不将用户输入插入子进程命令
• 失败时清理——多步操作中途失败时，清理部分状态

---

性能

• 优先编写清晰的代码，只有在有性能分析数据时才进行优化
• 对频繁实例化的数据类使用 __slots__
• 批量执行 I/O 操作（不要在循环中逐个读取文件）
• 对开销较大的资源进行延迟加载（首次使用时加载，而非导入时）
• 使用 asyncio.gather() 并发执行 I/O

---

日志记录

import colorlog

def get_logger(name: str) -> logging.Logger:
    handler = colorlog.StreamHandler()
    handler.setFormatter(colorlog.ColoredFormatter(
        "%(log_color)s%(levelname)-8s%(reset)s %(name)s: %(message)s"
    ))
    logger = logging.getLogger(name)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    return logger

用法：logger = get_logger(__name__)

---

Jinja2 模板

生成文本输出（HTML、配置、报告、markdown）时使用：

---

工具

工具	用途
uv	包管理、脚本执行
basedpyright	类型检查（严格模式）
ruff	严格的 lint + 格式化
pytest	测试
rusty-results	Result[T, E] 模式
typer	CLI 框架（首选）——始终配置对 -h 的支持（见下文）
argparse	仅用于只依赖标准库的脚本的 CLI
PySide6	GUI（无系统依赖）
httpx	HTTP（异步）
msgspec	外部数据验证与解析
platformdirs	各操作系统的用户配置/数据/缓存/日志目录（跨平台默认选择）
Jinja2	文本输出生成

始终启用 -h 帮助选项——typer 默认仅支持 --help：

app = typer.Typer(context_settings={"help_option_names": ["-h", "--help"]})

持续运行 uv run poe lint_full，不要只在最后运行。将其作为运行 basedpyright、Ruff 检查/格式化和自定义 linter 的主要检查流程；同时运行 uv run poe test 进行验证。

相关 myai 技能

• engineering-principles — 父级技能。与语言无关的理念：成功之道、快速失败、将错误处理作为控制流、测试理念、架构分离。
• python-reliable-subprocess-handling — 深入讲解上述异步子进程规则：安全捕获超时输出、孤立子进程及管道、测试替身。
• architecting-changes — 用于编写代码前的架构决策。加载 architecting-python-changes 以获取 Python 专属扩展。
• code-simplification — 用于重构 Python 代码，使其更清晰。