Python Engineering Excellence
Python卓越工程实践
This skill provides comprehensive Python engineering guidelines for modern Python development. Use this when writing or reviewing Python code for production systems, CLI tools, and AI agents.
本指南为现代Python开发提供全面的工程规范。适用于编写或审查生产系统、CLI工具及AI Agent的Python代码场景。
The Zen of Python (Selected)
Python之禅（精选）
- Explicit is better than implicit - Make intentions clear
- Simple is better than complex - Favor straightforward solutions
- Readability counts - Code is read more than written
- Errors should never pass silently - Handle or propagate errors explicitly
- There should be one obvious way to do it - Follow established patterns
- 显式优于隐式 - 清晰表达意图
- 简单优于复杂 - 优先选择直接的解决方案
- 可读性至关重要 - 代码被阅读的次数远多于编写次数
- 错误绝不应静默传递 - 显式处理或传播错误
- 总有一种显而易见的方式去做 - 遵循既定模式
- Pass dependencies explicitly, avoid global state
- Favor composition over inheritance
- Keep functions small and focused
- Make side effects obvious
- 显式传递依赖，避免全局状态
- 优先组合而非继承
- 保持函数小巧且聚焦单一职责
- 让副作用清晰可见
project/
src/
mypackage/
__init__.py
core.py
cli.py
agents/
__init__.py
graph.py
tests/
__init__.py
test_core.py
conftest.py
pyproject.toml
README.md
Guidelines:
- Use layout to avoid import confusion
- One package per project under
- Tests mirror source structure
- All configuration in
project/
src/
mypackage/
__init__.py
core.py
cli.py
agents/
__init__.py
graph.py
tests/
__init__.py
test_core.py
conftest.py
pyproject.toml
README.md
规范：
- 使用布局避免导入混淆
- 下每个项目对应一个包
- 测试目录镜像源码结构
- 所有配置统一放在中
pyproject.toml (uv)
pyproject.toml（uv）
toml
[project]
name = "mypackage"
version = "0.1.0"
description = "My package description"
requires-python = ">=3.11"
dependencies = [
"typer>=0.9.0",
"pydantic>=2.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23",
"mypy>=1.8",
]
[project.scripts]
mycli = "mypackage.cli:app"
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]
[tool.mypy]
python_version = "3.11"
warn_return_any = true
warn_unused_ignores = true
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
toml
[project]
name = "mypackage"
version = "0.1.0"
description = "My package description"
requires-python = ">=3.11"
dependencies = [
"typer>=0.9.0",
"pydantic>=2.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23",
"mypy>=1.8",
]
[project.scripts]
mycli = "mypackage.cli:app"
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]
[tool.mypy]
python_version = "3.11"
warn_return_any = true
warn_unused_ignores = true
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
module_name # modules: lowercase_with_underscores
ClassName # classes: CamelCase
function_name # functions: lowercase_with_underscores
variable_name # variables: lowercase_with_underscores
CONSTANT_NAME # constants: UPPERCASE_WITH_UNDERSCORES
_private_var # private: leading underscore
__mangled # name mangling: double underscore
module_name # modules: lowercase_with_underscores
ClassName # classes: CamelCase
function_name # functions: lowercase_with_underscores
variable_name # variables: lowercase_with_underscores
CONSTANT_NAME # constants: UPPERCASE_WITH_UNDERSCORES
_private_var # private: leading underscore
__mangled # name mangling: double underscore
ClassName() # Don't use for functions
functionName # Don't use camelCase for functions
VariableName # Don't use CamelCase for variables
ClassName() # Don't use for functions
functionName # Don't use camelCase for functions
VariableName # Don't use CamelCase for variables
Good - descriptive names
Good - descriptive names
def calculate_total_price(items: list[Item]) -> Decimal:
...
user_count = len(users)
is_valid = check_validity(data)
def calculate_total_price(items: list[Item]) -> Decimal:
...
user_count = len(users)
is_valid = check_validity(data)
Good - short names for small scopes
Good - short names for small scopes
for i, item in enumerate(items):
process(item)
for i, item in enumerate(items):
process(item)
Bad - single letters for large scopes
Bad - single letters for large scopes
def process(d): # What is d?
...
def process(d): # What is d?
...
Good - question-like predicates
Good - question-like predicates
is_active = True
has_permission = user.can_edit
should_retry = attempt < max_retries
is_active = True
has_permission = user.can_edit
should_retry = attempt < max_retries
active = True # Unclear if boolean
permission = True # Noun, not predicate
active = True # Unclear if boolean
permission = True # Noun, not predicate
Use
statements for packages and modules, not for individual classes or functions (except from
and
).
使用
语句导入包和模块，而非单个类或函数（
和
除外）。
Good - grouped and sorted (ruff handles this)
Good - grouped and sorted (ruff handles this)
from future import annotations
import asyncio
import os
from pathlib import Path
import httpx
from pydantic import BaseModel
from mypackage.core import process
from mypackage.models import User
from future import annotations
import asyncio
import os
from pathlib import Path
import httpx
from pydantic import BaseModel
from mypackage.core import process
from mypackage.models import User
Good - import modules, not individual items
Good - import modules, not individual items
from sound.effects import echo
echo.EchoFilter(...)
from sound.effects import echo
echo.EchoFilter(...)
Good - typing symbols can be imported directly
Good - typing symbols can be imported directly
from typing import Any, TypeVar
from collections.abc import Sequence, Mapping
from typing import Any, TypeVar
from collections.abc import Sequence, Mapping
Good - use aliases when needed
Good - use aliases when needed
import pandas as pd # Standard abbreviations OK
from mypackage.submodule import very_long_module as vlm
import pandas as pd # Standard abbreviations OK
from mypackage.submodule import very_long_module as vlm
Bad - importing individual classes (non-typing)
Bad - importing individual classes (non-typing)
from sound.effects.echo import EchoFilter
from sound.effects.echo import EchoFilter
Bad - ungrouped, relative imports in library code
Bad - ungrouped, relative imports in library code
from .core import process # Avoid relative imports
import os, sys # Never multiple imports on one line
from .core import process # Avoid relative imports
import os, sys # Never multiple imports on one line
Good - clear all for public API
Good - clear all for public API
all = ["Client", "process_data", "DataError"]
class DataError(Exception):
"""Raised when data processing fails."""
class Client:
"""HTTP client for the service."""
...
def process_data(data: bytes) -> dict:
"""Process raw data into structured format."""
...
all = ["Client", "process_data", "DataError"]
class DataError(Exception):
"""Raised when data processing fails."""
class Client:
"""HTTP client for the service."""
...
def process_data(data: bytes) -> dict:
"""Process raw data into structured format."""
...
Private helpers below public API
Private helpers below public API
def _validate(data: bytes) -> bool:
...
def _validate(data: bytes) -> bool:
...
python
class Service:
"""Service for processing requests."""
def __init__(self, client: Client, config: Config) -> None:
self._client = client
self._config = config
# Public methods first
def process(self, request: Request) -> Response:
data = self._fetch(request)
return self._transform(data)
# Private methods after
def _fetch(self, request: Request) -> bytes:
...
def _transform(self, data: bytes) -> Response:
...
python
class Service:
"""Service for processing requests."""
def __init__(self, client: Client, config: Config) -> None:
self._client = client
self._config = config
# Public methods first
def process(self, request: Request) -> Response:
data = self._fetch(request)
return self._transform(data)
# Private methods after
def _fetch(self, request: Request) -> bytes:
...
def _transform(self, data: bytes) -> Response:
...
Good - custom exceptions with context
Good - custom exceptions with context
class ProcessingError(Exception):
"""Raised when processing fails."""
def __init__(self, message: str, item_id: str) -> None:
super().__init__(message)
self.item_id = item_id
class ProcessingError(Exception):
"""Raised when processing fails."""
def __init__(self, message: str, item_id: str) -> None:
super().__init__(message)
self.item_id = item_id
Good - raise with context
Good - raise with context
def process(item: Item) -> Result:
try:
return transform(item)
except ValueError as e:
raise ProcessingError(f"failed to transform: {e}", item.id) from e
def process(item: Item) -> Result:
try:
return transform(item)
except ValueError as e:
raise ProcessingError(f"failed to transform: {e}", item.id) from e
Good - use built-in exceptions appropriately
Good - use built-in exceptions appropriately
def set_age(age: int) -> None:
if age < 0:
raise ValueError("age must be non-negative")
def set_age(age: int) -> None:
if age < 0:
raise ValueError("age must be non-negative")
Bad - bare except (catches KeyboardInterrupt, SystemExit)
Bad - bare except (catches KeyboardInterrupt, SystemExit)
try:
process(item)
except: # Never do this
pass
try:
process(item)
except: # Never do this
pass
Bad - catching Exception without re-raising
Bad - catching Exception without re-raising
try:
process(item)
except Exception:
pass # Silently swallowing errors
try:
process(item)
except Exception:
pass # Silently swallowing errors
OK - catching Exception only if re-raising or at isolation point
OK - catching Exception only if re-raising or at isolation point
try:
process(item)
except Exception:
logger.exception("Processing failed")
raise # Re-raise after logging
try:
process(item)
except Exception:
logger.exception("Processing failed")
raise # Re-raise after logging
Do not use
for validation or preconditions—use explicit conditionals:
不要使用
进行验证或前置条件检查——使用显式的条件判断：
Bad - assert can be disabled with -O flag
Bad - assert can be disabled with -O flag
def process(data: bytes) -> dict:
assert data, "data required" # Skipped in optimized mode!
return parse(data)
def process(data: bytes) -> dict:
assert data, "data required" # Skipped in optimized mode!
return parse(data)
Good - explicit validation
Good - explicit validation
def process(data: bytes) -> dict:
if not data:
raise ValueError("data required")
return parse(data)
def process(data: bytes) -> dict:
if not data:
raise ValueError("data required")
return parse(data)
OK - assert in tests (pytest)
OK - assert in tests (pytest)
def test_process():
result = process(b"test")
assert result["status"] == "ok"
def test_process():
result = process(b"test")
assert result["status"] == "ok"
Good - use context managers for cleanup
Good - use context managers for cleanup
from contextlib import contextmanager
@contextmanager
def managed_connection(dsn: str):
conn = connect(dsn)
try:
yield conn
finally:
conn.close()
from contextlib import contextmanager
@contextmanager
def managed_connection(dsn: str):
conn = connect(dsn)
try:
yield conn
finally:
conn.close()
with managed_connection(dsn) as conn:
conn.execute(query)
with managed_connection(dsn) as conn:
conn.execute(query)
Good - async context manager
Good - async context manager
from contextlib import asynccontextmanager
@asynccontextmanager
async def managed_session():
session = aiohttp.ClientSession()
try:
yield session
finally:
await session.close()
from contextlib import asynccontextmanager
@asynccontextmanager
async def managed_session():
session = aiohttp.ClientSession()
try:
yield session
finally:
await session.close()
Good - early returns reduce nesting
Good - early returns reduce nesting
def process_user(user: User | None) -> Result:
if user is None:
raise ValueError("user required")
if not user.is_active:
return Result.inactive()
if not user.has_permission("process"):
return Result.forbidden()
# Main logic at base indentation
return do_processing(user)
def process_user(user: User | None) -> Result:
if user is None:
raise ValueError("user required")
if not user.is_active:
return Result.inactive()
if not user.has_permission("process"):
return Result.forbidden()
# Main logic at base indentation
return do_processing(user)
Bad - deeply nested
Bad - deeply nested
def process_user(user: User | None) -> Result:
if user is not None:
if user.is_active:
if user.has_permission("process"):
return do_processing(user)
else:
return Result.forbidden()
else:
return Result.inactive()
else:
raise ValueError("user required")
def process_user(user: User | None) -> Result:
if user is not None:
if user.is_active:
if user.has_permission("process"):
return do_processing(user)
else:
return Result.forbidden()
else:
return Result.inactive()
else:
raise ValueError("user required")
Good - async function
Good - async function
async def fetch_data(url: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.json()
async def fetch_data(url: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.json()
Good - gather for concurrent operations
Good - gather for concurrent operations
async def fetch_all(urls: list[str]) -> list[dict]:
async with httpx.AsyncClient() as client:
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks)
return [r.json() for r in responses]
async def fetch_all(urls: list[str]) -> list[dict]:
async with httpx.AsyncClient() as client:
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks)
return [r.json() for r in responses]
Bad - sequential when concurrent is possible
Bad - sequential when concurrent is possible
async def fetch_all_slow(urls: list[str]) -> list[dict]:
results = []
for url in urls:
data = await fetch_data(url) # Sequential!
results.append(data)
return results
async def fetch_all_slow(urls: list[str]) -> list[dict]:
results = []
for url in urls:
data = await fetch_data(url) # Sequential!
results.append(data)
return results
Good - structured concurrency with TaskGroup (Python 3.11+)
Good - structured concurrency with TaskGroup (Python 3.11+)
async def process_items(items: list[Item]) -> list[Result]:
results = []
async with asyncio.TaskGroup() as tg:
for item in items:
task = tg.create_task(process_item(item))
results.append(task)
return [t.result() for t in results]
async def process_items(items: list[Item]) -> list[Result]:
results = []
async with asyncio.TaskGroup() as tg:
for item in items:
task = tg.create_task(process_item(item))
results.append(task)
return [t.result() for t in results]
Good - timeout handling
Good - timeout handling
async def fetch_with_timeout(url: str, timeout: float = 30.0) -> dict:
async with asyncio.timeout(timeout):
return await fetch_data(url)
async def fetch_with_timeout(url: str, timeout: float = 30.0) -> dict:
async with asyncio.timeout(timeout):
return await fetch_data(url)
Good - async generators
Good - async generators
async def stream_results(query: str):
async with get_connection() as conn:
async for row in conn.execute(query):
yield process_row(row)
async def stream_results(query: str):
async with get_connection() as conn:
async for row in conn.execute(query):
yield process_row(row)
async for result in stream_results(query):
handle(result)
async for result in stream_results(query):
handle(result)
Gradual Typing Approach
渐进式类型化方法
Type hints encouraged but not enforced. Prioritize:
- Public API functions and methods
- Function signatures (args + return)
- Complex data structures
- Code that benefits from IDE support
鼓励使用类型提示但不强制。优先级如下：
- 公共API函数和方法
- 函数签名（参数+返回值）
- 复杂数据结构
- 能从IDE支持中获益的代码
Good - typed public API
Good - typed public API
def process_items(items: list[Item], *, strict: bool = False) -> list[Result]:
"""Process items and return results."""
...
def process_items(items: list[Item], *, strict: bool = False) -> list[Result]:
"""Process items and return results."""
...
OK - internal helper without full typing
OK - internal helper without full typing
def _transform(data):
# Complex internal logic
...
def _transform(data):
# Complex internal logic
...
python
from typing import TypeVar, Protocol, Callable
from collections.abc import Iterator, Sequence
python
from typing import TypeVar, Protocol, Callable
from collections.abc import Iterator, Sequence
Generic types
Generic types
T = TypeVar("T")
def first(items: Sequence[T]) -> T | None:
return items[0] if items else None
T = TypeVar("T")
def first(items: Sequence[T]) -> T | None:
return items[0] if items else None
Protocols for structural typing
Protocols for structural typing
class Processor(Protocol):
def process(self, data: bytes) -> dict: ...
def run(processor: Processor, data: bytes) -> dict:
return processor.process(data)
class Processor(Protocol):
def process(self, data: bytes) -> dict: ...
def run(processor: Processor, data: bytes) -> dict:
return processor.process(data)
Callable types
Callable types
Handler = Callable[[Request], Response]
def with_logging(handler: Handler) -> Handler:
def wrapper(request: Request) -> Response:
log(request)
return handler(request)
return wrapper
Handler = Callable[[Request], Response]
def with_logging(handler: Handler) -> Handler:
def wrapper(request: Request) -> Response:
log(request)
return handler(request)
return wrapper
Union and Optional (use | syntax in 3.10+)
Union and Optional (use | syntax in 3.10+)
def find_user(user_id: str) -> User | None:
...
def find_user(user_id: str) -> User | None:
...
TypedDict for structured dicts
TypedDict for structured dicts
from typing import TypedDict
class UserData(TypedDict):
name: str
email: str
age: int | None
from typing import TypedDict
class UserData(TypedDict):
name: str
email: str
age: int | None
Good - type narrowing with isinstance
Good - type narrowing with isinstance
def process(value: str | int) -> str:
if isinstance(value, str):
return value.upper() # Type narrowed to str
return str(value)
def process(value: str | int) -> str:
if isinstance(value, str):
return value.upper() # Type narrowed to str
return str(value)
Good - assert for narrowing (use sparingly)
Good - assert for narrowing (use sparingly)
def process_user(user: User | None) -> str:
assert user is not None, "user required"
return user.name # Type narrowed to User
def process_user(user: User | None) -> str:
assert user is not None, "user required"
return user.name # Type narrowed to User
Line Length and Indentation
行长度与缩进
- Maximum 80 characters per line (URLs and long imports excepted)
- Use 4 spaces for indentation; never tabs
- Use implicit line continuation inside parentheses, brackets, braces
- 每行最多80个字符（URL和长导入除外）
- 使用4个空格缩进；绝不使用制表符
- 在括号、方括号、大括号内使用隐式换行
Good - implicit continuation with aligned elements
Good - implicit continuation with aligned elements
result = some_function(
argument_one,
argument_two,
argument_three,
)
result = some_function(
argument_one,
argument_two,
argument_three,
)
Good - hanging indent
Good - hanging indent
result = some_function(
argument_one, argument_two,
argument_three,
)
result = some_function(
argument_one, argument_two,
argument_three,
)
Bad - backslash continuation
Bad - backslash continuation
result = some_long_function_name()
+ another_function()
result = some_long_function_name()
+ another_function()
Good - parentheses for continuation
Good - parentheses for continuation
result = (
some_long_function_name()
+ another_function()
)
result = (
some_long_function_name()
+ another_function()
)
spam(ham[1], {eggs: 2})
x = 1
dict["key"] = list[index]
def func(default: str = "value") -> None: ...
spam(ham[1], {eggs: 2})
x = 1
dict["key"] = list[index]
def func(default: str = "value") -> None: ...
Bad - spaces inside brackets
Bad - spaces inside brackets
spam( ham[ 1 ], { eggs: 2 } )
spam( ham[ 1 ], { eggs: 2 } )
Bad - space before bracket
Bad - space before bracket
spam (ham[1])
dict ["key"]
spam (ham[1])
dict ["key"]
Good - break at highest syntactic level
Good - break at highest syntactic level
if (
condition_one
and condition_two
and condition_three
):
do_something()
if (
condition_one
and condition_two
and condition_three
):
do_something()
Bad - break in middle of expression
Bad - break in middle of expression
if (condition_one and
condition_two):
do_something()
if (condition_one and
condition_two):
do_something()
- Two blank lines between top-level definitions (functions, classes)
- One blank line between method definitions
- No blank line after line
- 顶层定义（函数、类）之间保留两个空行
- 方法定义之间保留一个空行
- 行后不要加空行
Use comprehensions for simple transformations. Avoid complex comprehensions.
Good - simple comprehension
Good - simple comprehension
squares = [x * x for x in range(10)]
evens = {x for x in numbers if x % 2 == 0}
mapping = {k: v for k, v in pairs}
squares = [x * x for x in range(10)]
evens = {x for x in numbers if x % 2 == 0}
mapping = {k: v for k, v in pairs}
Good - generator for large data
Good - generator for large data
total = sum(x * x for x in range(1000000))
total = sum(x * x for x in range(1000000))
Bad - multiple for clauses (hard to read)
Bad - multiple for clauses (hard to read)
result = [
(x, y, z)
for x in range(5)
for y in range(5)
for z in range(5)
if x != y
]
result = [
(x, y, z)
for x in range(5)
for y in range(5)
for z in range(5)
if x != y
]
Good - use nested loops instead
Good - use nested loops instead
result = []
for x in range(5):
for y in range(5):
for z in range(5):
if x != y:
result.append((x, y, z))
result = []
for x in range(5):
for y in range(5):
for z in range(5):
if x != y:
result.append((x, y, z))
Bad - complex conditions in comprehension
Bad - complex conditions in comprehension
result = [transform(x) for x in data if validate(x) and x.enabled and x.value > 0]
result = [transform(x) for x in data if validate(x) and x.enabled and x.value > 0]
Good - extract to function or use loop
Good - extract to function or use loop
def should_include(x):
return validate(x) and x.enabled and x.value > 0
result = [transform(x) for x in data if should_include(x)]
def should_include(x):
return validate(x) and x.enabled and x.value > 0
result = [transform(x) for x in data if should_include(x)]
Use f-strings for interpolation. For logging, use
format with pattern strings.
使用f-string进行插值。日志记录时，使用
格式搭配模式字符串。
Good - f-strings for general use
Good - f-strings for general use
message = f"Processing {item.name} (id={item.id})"
message = f"Processing {item.name} (id={item.id})"
Good - logging with % patterns (deferred formatting)
Good - logging with % patterns (deferred formatting)
logger.info("Processing %s (id=%s)", item.name, item.id)
logger.info("Processing %s (id=%s)", item.name, item.id)
Bad - f-strings in logging (always formatted even if not logged)
Bad - f-strings in logging (always formatted even if not logged)
logger.debug(f"Data: {expensive_repr(data)}")
logger.debug(f"Data: {expensive_repr(data)}")
Good - join for concatenation in loops
Good - join for concatenation in loops
parts = []
for item in items:
parts.append(str(item))
result = ", ".join(parts)
parts = []
for item in items:
parts.append(str(item))
result = ", ".join(parts)
result = ", ".join(str(item) for item in items)
result = ", ".join(str(item) for item in items)
Bad - += concatenation in loop
Bad - += concatenation in loop
result = ""
for item in items:
result += str(item) + ", " # Creates many intermediate strings
result = ""
for item in items:
result += str(item) + ", " # Creates many intermediate strings
Good - textwrap.dedent for indented multiline
Good - textwrap.dedent for indented multiline
import textwrap
long_string = textwrap.dedent("""
First line
Second line
Third line
""")
import textwrap
long_string = textwrap.dedent("""
First line
Second line
Third line
""")
Good - implicit concatenation
Good - implicit concatenation
message = (
"This is a very long message that needs "
"to be split across multiple lines for "
"readability purposes."
)
message = (
"This is a very long message that needs "
"to be split across multiple lines for "
"readability purposes."
)
Pydantic Settings
Pydantic Settings
python
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_prefix="MYAPP_",
env_file=".env",
)
database_url: str
api_key: str
debug: bool = False
max_workers: int = 4
python
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_prefix="MYAPP_",
env_file=".env",
)
database_url: str
api_key: str
debug: bool = False
max_workers: int = 4
settings = Settings() # Loads from env vars / .env
settings = Settings() # Loads from env vars / .env
CLI Arguments with Typer
CLI参数（Typer）
python
import typer
from typing import Annotated
app = typer.Typer()
@app.command()
def main(
input_file: Annotated[Path, typer.Argument(help="Input file path")],
output: Annotated[Path, typer.Option("--output", "-o")] = Path("out.json"),
verbose: Annotated[bool, typer.Option("--verbose", "-v")] = False,
) -> None:
"""Process input file and write results."""
if verbose:
print(f"Processing {input_file}")
result = process(input_file)
output.write_text(json.dumps(result))
if __name__ == "__main__":
app()
python
import typer
from typing import Annotated
app = typer.Typer()
@app.command()
def main(
input_file: Annotated[Path, typer.Argument(help="Input file path")],
output: Annotated[Path, typer.Option("--output", "-o")] = Path("out.json"),
verbose: Annotated[bool, typer.Option("--verbose", "-v")] = False,
) -> None:
"""Process input file and write results."""
if verbose:
print(f"Processing {input_file}")
result = process(input_file)
output.write_text(json.dumps(result))
if __name__ == "__main__":
app()
tests/test_core.py
tests/test_core.py
import pytest
from mypackage.core import process, ProcessingError
def test_process_valid_input():
result = process(b"valid data")
assert result["status"] == "ok"
def test_process_empty_input():
with pytest.raises(ProcessingError) as exc_info:
process(b"")
assert "empty" in str(exc_info.value)
import pytest
from mypackage.core import process, ProcessingError
def test_process_valid_input():
result = process(b"valid data")
assert result["status"] == "ok"
def test_process_empty_input():
with pytest.raises(ProcessingError) as exc_info:
process(b"")
assert "empty" in str(exc_info.value)
tests/conftest.py
tests/conftest.py
import pytest
from mypackage import Client, Config
@pytest.fixture
def config() -> Config:
return Config(api_url="
http://test", timeout=1.0)
@pytest.fixture
def client(config: Config) -> Client:
return Client(config)
import pytest
from mypackage import Client, Config
@pytest.fixture
def config() -> Config:
return Config(api_url="
http://test", timeout=1.0)
@pytest.fixture
def client(config: Config) -> Client:
return Client(config)
Fixture with cleanup
Fixture with cleanup
@pytest.fixture
def temp_db(tmp_path):
db_path = tmp_path / "test.db"
db = create_database(db_path)
yield db
db.close()
@pytest.fixture
def temp_db(tmp_path):
db_path = tmp_path / "test.db"
db = create_database(db_path)
yield db
db.close()
Async fixtures
Async fixtures
@pytest.fixture
async def async_client():
async with httpx.AsyncClient() as client:
yield client
@pytest.fixture
async def async_client():
async with httpx.AsyncClient() as client:
yield client
python
@pytest.mark.parametrize(
"input_data,expected",
[
(b"hello", {"word": "hello", "length": 5}),
(b"world", {"word": "world", "length": 5}),
(b"", None),
],
ids=["hello", "world", "empty"],
)
def test_parse(input_data: bytes, expected: dict | None):
result = parse(input_data)
assert result == expected
python
@pytest.mark.parametrize(
"input_data,expected",
[
(b"hello", {"word": "hello", "length": 5}),
(b"world", {"word": "world", "length": 5}),
(b"", None),
],
ids=["hello", "world", "empty"],
)
def test_parse(input_data: bytes, expected: dict | None):
result = parse(input_data)
assert result == expected
python
from unittest.mock import Mock, AsyncMock, patch
def test_with_mock():
mock_client = Mock()
mock_client.fetch.return_value = {"data": "test"}
service = Service(client=mock_client)
result = service.process()
mock_client.fetch.assert_called_once()
assert result == {"data": "test"}
python
from unittest.mock import Mock, AsyncMock, patch
def test_with_mock():
mock_client = Mock()
mock_client.fetch.return_value = {"data": "test"}
service = Service(client=mock_client)
result = service.process()
mock_client.fetch.assert_called_once()
assert result == {"data": "test"}
async def test_async_service():
mock_client = AsyncMock()
mock_client.fetch.return_value = {"data": "test"}
service = AsyncService(client=mock_client)
result = await service.process()
assert result == {"data": "test"}
async def test_async_service():
mock_client = AsyncMock()
mock_client.fetch.return_value = {"data": "test"}
service = AsyncService(client=mock_client)
result = await service.process()
assert result == {"data": "test"}
@patch("mypackage.core.external_api")
def test_with_patch(mock_api):
mock_api.return_value = "mocked"
result = function_using_api()
assert result == "mocked"
@patch("mypackage.core.external_api")
def test_with_patch(mock_api):
mock_api.return_value = "mocked"
result = function_using_api()
assert result == "mocked"
Typer Application Structure
Typer应用结构
src/mypackage/cli.py
src/mypackage/cli.py
import typer
from rich.console import Console
app = typer.Typer(help="My CLI application")
console = Console()
@app.command()
def init(
name: str,
force: bool = typer.Option(False, "--force", "-f", help="Overwrite existing"),
) -> None:
"""Initialize a new project."""
if Path(name).exists() and not force:
console.print(f"[red]Error:[/red] {name} already exists")
raise typer.Exit(1)
create_project(name)
console.print(f"[green]Created[/green] {name}")
@app.command()
def run(
config: Path = typer.Option(Path("config.yaml"), "--config", "-c"),
) -> None:
"""Run the application."""
settings = load_config(config)
execute(settings)
if name == "main":
app()
import typer
from rich.console import Console
app = typer.Typer(help="My CLI application")
console = Console()
@app.command()
def init(
name: str,
force: bool = typer.Option(False, "--force", "-f", help="Overwrite existing"),
) -> None:
"""Initialize a new project."""
if Path(name).exists() and not force:
console.print(f"[red]Error:[/red] {name} already exists")
raise typer.Exit(1)
create_project(name)
console.print(f"[green]Created[/green] {name}")
@app.command()
def run(
config: Path = typer.Option(Path("config.yaml"), "--config", "-c"),
) -> None:
"""Run the application."""
settings = load_config(config)
execute(settings)
if name == "main":
app()
Main app with subcommands
Main app with subcommands
app = typer.Typer()
db_app = typer.Typer(help="Database commands")
app.add_typer(db_app, name="db")
@db_app.command("migrate")
def db_migrate():
"""Run database migrations."""
...
@db_app.command("seed")
def db_seed():
"""Seed database with test data."""
...
app = typer.Typer()
db_app = typer.Typer(help="Database commands")
app.add_typer(db_app, name="db")
@db_app.command("migrate")
def db_migrate():
"""Run database migrations."""
...
@db_app.command("seed")
def db_seed():
"""Seed database with test data."""
...
Usage: mycli db migrate
Usage: mycli db migrate
AI Agents / LangGraph
AI Agent / LangGraph
python
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
context: dict
next_step: str | None
def create_agent() -> StateGraph:
graph = StateGraph(AgentState)
graph.add_node("process", process_node)
graph.add_node("decide", decision_node)
graph.add_node("execute", execute_node)
graph.add_edge("process", "decide")
graph.add_conditional_edges(
"decide",
route_decision,
{"execute": "execute", "end": "__end__"},
)
graph.add_edge("execute", "process")
graph.set_entry_point("process")
return graph.compile()
python
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
context: dict
next_step: str | None
def create_agent() -> StateGraph:
graph = StateGraph(AgentState)
graph.add_node("process", process_node)
graph.add_node("decide", decision_node)
graph.add_node("execute", execute_node)
graph.add_edge("process", "decide")
graph.add_conditional_edges(
"decide",
route_decision,
{"execute": "execute", "end": "__end__"},
)
graph.add_edge("execute", "process")
graph.set_entry_point("process")
return graph.compile()
python
from langchain_core.tools import tool
@tool
def search_database(query: str) -> list[dict]:
"""Search the database for matching records.
Args:
query: Search query string
Returns:
List of matching records
"""
return db.search(query)
@tool
async def fetch_url(url: str) -> str:
"""Fetch content from a URL.
Args:
url: URL to fetch
Returns:
Page content as text
"""
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.text
python
from langchain_core.tools import tool
@tool
def search_database(query: str) -> list[dict]:
"""Search the database for matching records.
Args:
query: Search query string
Returns:
List of matching records
"""
return db.search(query)
@tool
async def fetch_url(url: str) -> str:
"""Fetch content from a URL.
Args:
url: URL to fetch
Returns:
Page content as text
"""
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.text
python
from langchain_core.messages import HumanMessage, AIMessage
async def process_node(state: AgentState) -> AgentState:
"""Process incoming messages and update context."""
last_message = state["messages"][-1]
if isinstance(last_message, HumanMessage):
# Process user input
context = await analyze_message(last_message.content)
return {"context": context}
return {}
def decision_node(state: AgentState) -> AgentState:
"""Decide next action based on context."""
context = state["context"]
if context.get("needs_execution"):
return {"next_step": "execute"}
return {"next_step": "end"}
def route_decision(state: AgentState) -> str:
"""Route based on decision."""
return state.get("next_step", "end")
python
from langchain_core.messages import HumanMessage, AIMessage
async def process_node(state: AgentState) -> AgentState:
"""Process incoming messages and update context."""
last_message = state["messages"][-1]
if isinstance(last_message, HumanMessage):
# Process user input
context = await analyze_message(last_message.content)
return {"context": context}
return {}
def decision_node(state: AgentState) -> AgentState:
"""Decide next action based on context."""
context = state["context"]
if context.get("needs_execution"):
return {"next_step": "execute"}
return {"next_step": "end"}
def route_decision(state: AgentState) -> str:
"""Route based on decision."""
return state.get("next_step", "end")
Constructor Injection
构造函数注入
Good - dependencies passed explicitly
Good - dependencies passed explicitly
class UserService:
def init(self, db: Database, cache: Cache, logger: Logger) -> None:
self._db = db
self._cache = cache
self._logger = logger
def get_user(self, user_id: str) -> User | None:
if cached := self._cache.get(user_id):
return cached
user = self._db.find_user(user_id)
if user:
self._cache.set(user_id, user)
return user
class UserService:
def init(self, db: Database, cache: Cache, logger: Logger) -> None:
self._db = db
self._cache = cache
self._logger = logger
def get_user(self, user_id: str) -> User | None:
if cached := self._cache.get(user_id):
return cached
user = self._db.find_user(user_id)
if user:
self._cache.set(user_id, user)
return user
Bad - hidden dependencies
Bad - hidden dependencies
class UserService:
def get_user(self, user_id: str) -> User | None:
return global_db.find_user(user_id) # Hidden dependency
class UserService:
def get_user(self, user_id: str) -> User | None:
return global_db.find_user(user_id) # Hidden dependency
python
def create_service(config: Config) -> Service:
"""Create service with all dependencies configured."""
db = Database(config.database_url)
cache = RedisCache(config.redis_url)
logger = setup_logger(config.log_level)
return Service(db=db, cache=cache, logger=logger)
python
def create_service(config: Config) -> Service:
"""Create service with all dependencies configured."""
db = Database(config.database_url)
cache = RedisCache(config.redis_url)
logger = setup_logger(config.log_level)
return Service(db=db, cache=cache, logger=logger)
def create_test_service() -> Service:
return Service(
db=InMemoryDatabase(),
cache=DictCache(),
logger=NullLogger(),
)
def create_test_service() -> Service:
return Service(
db=InMemoryDatabase(),
cache=DictCache(),
logger=NullLogger(),
)
Always guard module-level code to prevent execution on import:
Good - guarded entry point
Good - guarded entry point
def main() -> None:
"""Application entry point."""
config = load_config()
result = process(config)
print(result)
if name == "main":
main()
def main() -> None:
"""Application entry point."""
config = load_config()
result = process(config)
print(result)
if name == "main":
main()
With CLI framework (typer)
With CLI framework (typer)
import typer
app = typer.Typer()
@app.command()
def main(config: Path = typer.Option(...)) -> None:
"""Process with configuration."""
...
if name == "main":
app()
import typer
app = typer.Typer()
@app.command()
def main(config: Path = typer.Option(...)) -> None:
"""Process with configuration."""
...
if name == "main":
app()
Anti-patterns to Avoid
需避免的反模式
Mutable Default Arguments
可变默认参数
Bad - mutable default shared across calls
Bad - mutable default shared across calls
def append_item(item, items=[]):
items.append(item)
return items
def append_item(item, items=[]):
items.append(item)
return items
Good - use None and create new list
Good - use None and create new list
def append_item(item, items=None):
if items is None:
items = []
items.append(item)
return items
def append_item(item, items=None):
if items is None:
items = []
items.append(item)
return items
Bad - catches everything including KeyboardInterrupt
Bad - catches everything including KeyboardInterrupt
try:
process()
except:
pass
try:
process()
except:
pass
Good - catch specific exceptions
Good - catch specific exceptions
try:
process()
except ProcessingError as e:
logger.error(f"Processing failed: {e}")
raise
try:
process()
except ProcessingError as e:
logger.error(f"Processing failed: {e}")
raise
Bad - pollutes namespace, unclear origins
Bad - pollutes namespace, unclear origins
Good - explicit imports
Good - explicit imports
from module import SpecificClass, specific_function
from module import SpecificClass, specific_function
Overusing Inheritance
过度使用继承
Bad - deep inheritance hierarchy
Bad - deep inheritance hierarchy
class Animal: ...
class Mammal(Animal): ...
class Dog(Mammal): ...
class GermanShepherd(Dog): ...
class Animal: ...
class Mammal(Animal): ...
class Dog(Mammal): ...
class GermanShepherd(Dog): ...
Good - composition and protocols
Good - composition and protocols
class Animal(Protocol):
def speak(self) -> str: ...
class Dog:
def init(self, breed: str) -> None:
self.breed = breed
def speak(self) -> str:
return "woof"
class Animal(Protocol):
def speak(self) -> str: ...
class Dog:
def init(self, breed: str) -> None:
self.breed = breed
def speak(self) -> str:
return "woof"
Bad - class doing too much
Bad - class doing too much
class Application:
def connect_database(self): ...
def send_email(self): ...
def process_payment(self): ...
def generate_report(self): ...
def authenticate_user(self): ...
class Application:
def connect_database(self): ...
def send_email(self): ...
def process_payment(self): ...
def generate_report(self): ...
def authenticate_user(self): ...
Good - single responsibility
Good - single responsibility
class DatabaseConnection: ...
class EmailService: ...
class PaymentProcessor: ...
class ReportGenerator: ...
class AuthService: ...
class DatabaseConnection: ...
class EmailService: ...
class PaymentProcessor: ...
class ReportGenerator: ...
class AuthService: ...
Avoid Power Features
避免过度使用高级特性
Avoid metaclasses, dynamic attribute access via
, bytecode manipulation, and reflection tricks. Use simpler alternatives.
避免元类、通过
进行动态属性访问、字节码操作以及反射技巧。使用更简单的替代方案。
Bad - metaclass for simple use case
Bad - metaclass for simple use case
class SingletonMeta(type):
_instances = {}
def call(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().call(*args, **kwargs)
return cls._instances[cls]
class SingletonMeta(type):
_instances = {}
def call(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().call(*args, **kwargs)
return cls._instances[cls]
Good - module-level instance or factory function
Good - module-level instance or factory function
_instance = None
def get_instance() -> Service:
global _instance
if _instance is None:
_instance = Service()
return _instance
_instance = None
def get_instance() -> Service:
global _instance
if _instance is None:
_instance = Service()
return _instance
Avoid staticmethod
避免staticmethod
Use module-level functions instead of
:
Bad - staticmethod
Bad - staticmethod
class StringUtils:
@staticmethod
def clean(text: str) -> str:
return text.strip().lower()
class StringUtils:
@staticmethod
def clean(text: str) -> str:
return text.strip().lower()
Good - module-level function
Good - module-level function
def clean_text(text: str) -> str:
return text.strip().lower()
def clean_text(text: str) -> str:
return text.strip().lower()
Boolean Evaluation Pitfalls
布尔值判断陷阱
Good - explicit None check
Good - explicit None check
if value is not None:
process(value)
if value is not None:
process(value)
Bad - falsy check when 0 or "" are valid
Bad - falsy check when 0 or "" are valid
if value: # Fails for value=0 or value=""
process(value)
if value: # Fails for value=0 or value=""
process(value)
Good - explicit comparison for sequences
Good - explicit comparison for sequences
if len(items) == 0:
return default
if len(items) == 0:
return default
OK - implicit boolean for sequences (when falsy means empty)
OK - implicit boolean for sequences (when falsy means empty)
if not items:
return default
if not items:
return default
python
def process_data(
data: bytes,
*,
encoding: str = "utf-8",
strict: bool = False,
) -> dict:
"""Process raw bytes into structured data.
Args:
data: Raw bytes to process
encoding: Text encoding to use
strict: If True, raise on invalid data
Returns:
Parsed data as dictionary
Raises:
ProcessingError: If data is invalid and strict=True
Example:
>>> result = process_data(b'{"key": "value"}')
>>> result["key"]
'value'
"""
...
python
def process_data(
data: bytes,
*,
encoding: str = "utf-8",
strict: bool = False,
) -> dict:
"""Process raw bytes into structured data.
Args:
data: Raw bytes to process
encoding: Text encoding to use
strict: If True, raise on invalid data
Returns:
Parsed data as dictionary
Raises:
ProcessingError: If data is invalid and strict=True
Example:
>>> result = process_data(b'{"key": "value"}')
>>> result["key"]
'value'
"""
...
python
class DataProcessor:
"""Process and transform data from various sources.
This class handles reading data from files or URLs,
validating the format, and transforming it into
the required output structure.
Attributes:
config: Processor configuration
stats: Processing statistics
Example:
>>> processor = DataProcessor(Config())
>>> result = processor.process(data)
"""
def __init__(self, config: Config) -> None:
"""Initialize processor with configuration.
Args:
config: Processor configuration object
"""
self.config = config
self.stats = Stats()
python
class DataProcessor:
"""Process and transform data from various sources.
This class handles reading data from files or URLs,
validating the format, and transforming it into
the required output structure.
Attributes:
config: Processor configuration
stats: Processing statistics
Example:
>>> processor = DataProcessor(Config())
>>> result = processor.process(data)
"""
def __init__(self, config: Config) -> None:
"""Initialize processor with configuration.
Args:
config: Processor configuration object
"""
self.config = config
self.stats = Stats()
Create new project
Create new project
uv init myproject
cd myproject
uv init myproject
cd myproject
Add dependencies
Add dependencies
uv add httpx pydantic typer
uv add httpx pydantic typer
Add dev dependencies
Add dev dependencies
uv add --dev pytest pytest-asyncio mypy ruff
uv add --dev pytest pytest-asyncio mypy ruff
Sync dependencies
Sync dependencies
Run commands in venv
Run commands in venv
uv run python script.py
uv run pytest
uv run mypy src/
uv run python script.py
uv run pytest
uv run mypy src/
Build package
Build package
Check for issues
Check for issues
Fix auto-fixable issues
Fix auto-fixable issues
uv run ruff check --fix .
uv run ruff check --fix .
Check formatting without changes
Check formatting without changes
uv run ruff format --check .
uv run ruff format --check .
Strict mode (if desired)
Strict mode (if desired)
uv run mypy --strict src/
uv run mypy --strict src/
Generate stub files
Generate stub files
uv run stubgen -p mypackage
uv run stubgen -p mypackage
Run all tests
Run all tests
Verbose with output
Verbose with output
Run specific test
Run specific test
uv run pytest tests/test_core.py::test_process
uv run pytest tests/test_core.py::test_process
Run with coverage
Run with coverage
uv run pytest --cov=mypackage --cov-report=html
uv run pytest --cov=mypackage --cov-report=html
Run async tests (with pytest-asyncio)
Run async tests (with pytest-asyncio)
uv run pytest # asyncio_mode = "auto" in pyproject.toml
uv run pytest # asyncio_mode = "auto" in pyproject.toml
Pre-commit Workflow
预提交工作流
2. Lint and fix
2. Lint and fix
uv run ruff check --fix .
uv run ruff check --fix .
3. Type check
3. Type check
5. All checks pass, ready to commit
5. All checks pass, ready to commit
Remember: These guidelines support writing clear, maintainable Python code. Adapt them to your specific context, but always favor readability, explicitness, and simplicity.
注意： 这些指南旨在帮助编写清晰、可维护的Python代码。可根据具体场景调整，但始终优先考虑可读性、显式性和简洁性。