python-design-patterns by wshobson/agents
npx skills add https://github.com/wshobson/agents --skill python-design-patterns运用基础设计原则编写可维护的 Python 代码。这些模式帮助你构建易于理解、测试和修改的系统。
选择可行的最简单方案。复杂性必须有具体的需求来证明其合理性。
每个单元应该只有一个变更的理由。将关注点分离到专注的组件中。
通过组合对象而非扩展类来构建行为。
等到有三个实例后再进行抽象。重复通常优于过早抽象。
# 简单优于巧妙
# 替代工厂/注册模式:
FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter}
def get_formatter(name: str) -> Formatter:
return FORMATTERS[name]()
在增加复杂性之前,先问:是否有更简单的解决方案可行?
# 过度设计:带注册功能的工厂
class OutputFormatterFactory:
_formatters: dict[str, type[Formatter]] = {}
@classmethod
def register(cls, name: str):
def decorator(formatter_cls):
cls._formatters[name] = formatter_cls
return formatter_cls
return decorator
@classmethod
def create(cls, name: str) -> Formatter:
return cls._formatters[name]()
@OutputFormatterFactory.register("json")
class JsonFormatter(Formatter):
...
# 简单方案:直接使用字典
FORMATTERS = {
"json": JsonFormatter,
"csv": CsvFormatter,
"xml": XmlFormatter,
}
def get_formatter(name: str) -> Formatter:
"""根据名称获取格式化器。"""
if name not in FORMATTERS:
raise ValueError(f"Unknown format: {name}")
return FORMATTERS[name]()
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
在此场景下,工厂模式增加了代码却没有增加价值。将模式留到它们能解决实际问题时再用。
每个类或函数应该只有一个变更的理由。
# 不好:处理器做所有事情
class UserHandler:
async def create_user(self, request: Request) -> Response:
# HTTP 解析
data = await request.json()
# 验证
if not data.get("email"):
return Response({"error": "email required"}, status=400)
# 数据库访问
user = await db.execute(
"INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *",
data["email"], data["name"]
)
# 响应格式化
return Response({"id": user.id, "email": user.email}, status=201)
# 好:分离关注点
class UserService:
"""仅包含业务逻辑。"""
def __init__(self, repo: UserRepository) -> None:
self._repo = repo
async def create_user(self, data: CreateUserInput) -> User:
# 这里只有业务规则
user = User(email=data.email, name=data.name)
return await self._repo.save(user)
class UserHandler:
"""仅处理 HTTP 相关事宜。"""
def __init__(self, service: UserService) -> None:
self._service = service
async def create_user(self, request: Request) -> Response:
data = CreateUserInput(**(await request.json()))
user = await self._service.create_user(data)
return Response(user.to_dict(), status=201)
现在,HTTP 的变更不会影响业务逻辑,反之亦然。
将代码组织成具有明确职责的不同层次。
┌─────────────────────────────────────────────────────┐
│ API 层(处理器) │
│ - 解析请求 │
│ - 调用服务 │
│ - 格式化响应 │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 服务层(业务逻辑) │
│ - 领域规则和验证 │
│ - 编排操作 │
│ - 尽可能使用纯函数 │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 仓库层(数据访问) │
│ - SQL 查询 │
│ - 外部 API 调用 │
│ - 缓存操作 │
└─────────────────────────────────────────────────────┘
每一层仅依赖于其下方的层:
# 仓库层:数据访问
class UserRepository:
async def get_by_id(self, user_id: str) -> User | None:
row = await self._db.fetchrow(
"SELECT * FROM users WHERE id = $1", user_id
)
return User(**row) if row else None
# 服务层:业务逻辑
class UserService:
def __init__(self, repo: UserRepository) -> None:
self._repo = repo
async def get_user(self, user_id: str) -> User:
user = await self._repo.get_by_id(user_id)
if user is None:
raise UserNotFoundError(user_id)
return user
# 处理器层:HTTP 相关事宜
@app.get("/users/{user_id}")
async def get_user(user_id: str) -> UserResponse:
user = await user_service.get_user(user_id)
return UserResponse.from_user(user)
通过组合对象而非继承来构建行为。
# 继承:僵化且难以测试
class EmailNotificationService(NotificationService):
def __init__(self):
super().__init__()
self._smtp = SmtpClient() # 难以模拟
def notify(self, user: User, message: str) -> None:
self._smtp.send(user.email, message)
# 组合:灵活且可测试
class NotificationService:
"""通过多种渠道发送通知。"""
def __init__(
self,
email_sender: EmailSender,
sms_sender: SmsSender | None = None,
push_sender: PushSender | None = None,
) -> None:
self._email = email_sender
self._sms = sms_sender
self._push = push_sender
async def notify(
self,
user: User,
message: str,
channels: set[str] | None = None,
) -> None:
channels = channels or {"email"}
if "email" in channels:
await self._email.send(user.email, message)
if "sms" in channels and self._sms and user.phone:
await self._sms.send(user.phone, message)
if "push" in channels and self._push and user.device_token:
await self._push.send(user.device_token, message)
# 使用假对象易于测试
service = NotificationService(
email_sender=FakeEmailSender(),
sms_sender=FakeSmsSender(),
)
等到有三个实例后再进行抽象。
# 两个相似的函数?先别抽象
def process_orders(orders: list[Order]) -> list[Result]:
results = []
for order in orders:
validated = validate_order(order)
result = process_validated_order(validated)
results.append(result)
return results
def process_returns(returns: list[Return]) -> list[Result]:
results = []
for ret in returns:
validated = validate_return(ret)
result = process_validated_return(validated)
results.append(result)
return results
# 它们看起来相似,但是等等!它们真的相同吗?
# 不同的验证、不同的处理、不同的错误...
# 重复通常优于错误的抽象
# 只有在出现第三个用例后,才考虑是否存在真正的模式
# 但即便如此,有时显式也比抽象更好
保持函数专注。在以下情况提取函数:
# 太长,多个关注点混杂
def process_order(order: Order) -> Result:
# 50 行验证...
# 30 行库存检查...
# 40 行支付处理...
# 20 行通知...
pass
# 更好:由专注的函数组合而成
def process_order(order: Order) -> Result:
"""通过完整工作流处理客户订单。"""
validate_order(order)
reserve_inventory(order)
payment_result = charge_payment(order)
send_confirmation(order, payment_result)
return Result(success=True, order_id=order.id)
通过构造函数传递依赖项以提高可测试性。
from typing import Protocol
class Logger(Protocol):
def info(self, msg: str, **kwargs) -> None: ...
def error(self, msg: str, **kwargs) -> None: ...
class Cache(Protocol):
async def get(self, key: str) -> str | None: ...
async def set(self, key: str, value: str, ttl: int) -> None: ...
class UserService:
"""具有注入依赖项的服务。"""
def __init__(
self,
repository: UserRepository,
cache: Cache,
logger: Logger,
) -> None:
self._repo = repository
self._cache = cache
self._logger = logger
async def get_user(self, user_id: str) -> User:
# 先检查缓存
cached = await self._cache.get(f"user:{user_id}")
if cached:
self._logger.info("Cache hit", user_id=user_id)
return User.from_json(cached)
# 从数据库获取
user = await self._repo.get_by_id(user_id)
if user:
await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300)
return user
# 生产环境
service = UserService(
repository=PostgresUserRepository(db),
cache=RedisCache(redis),
logger=StructlogLogger(),
)
# 测试环境
service = UserService(
repository=InMemoryUserRepository(),
cache=FakeCache(),
logger=NullLogger(),
)
不要暴露内部类型:
# 不好:将 ORM 模型泄露给 API
@app.get("/users/{id}")
def get_user(id: str) -> UserModel: # SQLAlchemy 模型
return db.query(UserModel).get(id)
# 好:使用响应模式
@app.get("/users/{id}")
def get_user(id: str) -> UserResponse:
user = db.query(UserModel).get(id)
return UserResponse.from_orm(user)
不要将 I/O 与业务逻辑混合:
# 不好:SQL 嵌入在业务逻辑中
def calculate_discount(user_id: str) -> float:
user = db.query("SELECT * FROM users WHERE id = ?", user_id)
orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
# 业务逻辑与数据访问混杂
# 好:仓库模式
def calculate_discount(user: User, order_history: list[Order]) -> float:
# 纯业务逻辑,易于测试
if len(order_history) > 10:
return 0.15
return 0.0
每周安装量
5.4K
仓库
GitHub 星标
32.2K
首次出现
2026 年 1 月 30 日
安全审计
安装于
opencode4.5K
gemini-cli4.5K
codex4.4K
github-copilot4.2K
kimi-cli3.9K
amp3.9K
Write maintainable Python code using fundamental design principles. These patterns help you build systems that are easy to understand, test, and modify.
Choose the simplest solution that works. Complexity must be justified by concrete requirements.
Each unit should have one reason to change. Separate concerns into focused components.
Build behavior by combining objects, not extending classes.
Wait until you have three instances before abstracting. Duplication is often better than premature abstraction.
# Simple beats clever
# Instead of a factory/registry pattern:
FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter}
def get_formatter(name: str) -> Formatter:
return FORMATTERS[name]()
Before adding complexity, ask: does a simpler solution work?
# Over-engineered: Factory with registration
class OutputFormatterFactory:
_formatters: dict[str, type[Formatter]] = {}
@classmethod
def register(cls, name: str):
def decorator(formatter_cls):
cls._formatters[name] = formatter_cls
return formatter_cls
return decorator
@classmethod
def create(cls, name: str) -> Formatter:
return cls._formatters[name]()
@OutputFormatterFactory.register("json")
class JsonFormatter(Formatter):
...
# Simple: Just use a dictionary
FORMATTERS = {
"json": JsonFormatter,
"csv": CsvFormatter,
"xml": XmlFormatter,
}
def get_formatter(name: str) -> Formatter:
"""Get formatter by name."""
if name not in FORMATTERS:
raise ValueError(f"Unknown format: {name}")
return FORMATTERS[name]()
The factory pattern adds code without adding value here. Save patterns for when they solve real problems.
Each class or function should have one reason to change.
# BAD: Handler does everything
class UserHandler:
async def create_user(self, request: Request) -> Response:
# HTTP parsing
data = await request.json()
# Validation
if not data.get("email"):
return Response({"error": "email required"}, status=400)
# Database access
user = await db.execute(
"INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *",
data["email"], data["name"]
)
# Response formatting
return Response({"id": user.id, "email": user.email}, status=201)
# GOOD: Separated concerns
class UserService:
"""Business logic only."""
def __init__(self, repo: UserRepository) -> None:
self._repo = repo
async def create_user(self, data: CreateUserInput) -> User:
# Only business rules here
user = User(email=data.email, name=data.name)
return await self._repo.save(user)
class UserHandler:
"""HTTP concerns only."""
def __init__(self, service: UserService) -> None:
self._service = service
async def create_user(self, request: Request) -> Response:
data = CreateUserInput(**(await request.json()))
user = await self._service.create_user(data)
return Response(user.to_dict(), status=201)
Now HTTP changes don't affect business logic, and vice versa.
Organize code into distinct layers with clear responsibilities.
┌─────────────────────────────────────────────────────┐
│ API Layer (handlers) │
│ - Parse requests │
│ - Call services │
│ - Format responses │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Service Layer (business logic) │
│ - Domain rules and validation │
│ - Orchestrate operations │
│ - Pure functions where possible │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Repository Layer (data access) │
│ - SQL queries │
│ - External API calls │
│ - Cache operations │
└─────────────────────────────────────────────────────┘
Each layer depends only on layers below it:
# Repository: Data access
class UserRepository:
async def get_by_id(self, user_id: str) -> User | None:
row = await self._db.fetchrow(
"SELECT * FROM users WHERE id = $1", user_id
)
return User(**row) if row else None
# Service: Business logic
class UserService:
def __init__(self, repo: UserRepository) -> None:
self._repo = repo
async def get_user(self, user_id: str) -> User:
user = await self._repo.get_by_id(user_id)
if user is None:
raise UserNotFoundError(user_id)
return user
# Handler: HTTP concerns
@app.get("/users/{user_id}")
async def get_user(user_id: str) -> UserResponse:
user = await user_service.get_user(user_id)
return UserResponse.from_user(user)
Build behavior by combining objects rather than inheriting.
# Inheritance: Rigid and hard to test
class EmailNotificationService(NotificationService):
def __init__(self):
super().__init__()
self._smtp = SmtpClient() # Hard to mock
def notify(self, user: User, message: str) -> None:
self._smtp.send(user.email, message)
# Composition: Flexible and testable
class NotificationService:
"""Send notifications via multiple channels."""
def __init__(
self,
email_sender: EmailSender,
sms_sender: SmsSender | None = None,
push_sender: PushSender | None = None,
) -> None:
self._email = email_sender
self._sms = sms_sender
self._push = push_sender
async def notify(
self,
user: User,
message: str,
channels: set[str] | None = None,
) -> None:
channels = channels or {"email"}
if "email" in channels:
await self._email.send(user.email, message)
if "sms" in channels and self._sms and user.phone:
await self._sms.send(user.phone, message)
if "push" in channels and self._push and user.device_token:
await self._push.send(user.device_token, message)
# Easy to test with fakes
service = NotificationService(
email_sender=FakeEmailSender(),
sms_sender=FakeSmsSender(),
)
Wait until you have three instances before abstracting.
# Two similar functions? Don't abstract yet
def process_orders(orders: list[Order]) -> list[Result]:
results = []
for order in orders:
validated = validate_order(order)
result = process_validated_order(validated)
results.append(result)
return results
def process_returns(returns: list[Return]) -> list[Result]:
results = []
for ret in returns:
validated = validate_return(ret)
result = process_validated_return(validated)
results.append(result)
return results
# These look similar, but wait! Are they actually the same?
# Different validation, different processing, different errors...
# Duplication is often better than the wrong abstraction
# Only after a third case, consider if there's a real pattern
# But even then, sometimes explicit is better than abstract
Keep functions focused. Extract when a function:
Exceeds 20-50 lines (varies by complexity)
Serves multiple distinct purposes
Has deeply nested logic (3+ levels)
def process_order(order: Order) -> Result: # 50 lines of validation... # 30 lines of inventory check... # 40 lines of payment processing... # 20 lines of notification... pass
def process_order(order: Order) -> Result: """Process a customer order through the complete workflow.""" validate_order(order) reserve_inventory(order) payment_result = charge_payment(order) send_confirmation(order, payment_result) return Result(success=True, order_id=order.id)
Pass dependencies through constructors for testability.
from typing import Protocol
class Logger(Protocol):
def info(self, msg: str, **kwargs) -> None: ...
def error(self, msg: str, **kwargs) -> None: ...
class Cache(Protocol):
async def get(self, key: str) -> str | None: ...
async def set(self, key: str, value: str, ttl: int) -> None: ...
class UserService:
"""Service with injected dependencies."""
def __init__(
self,
repository: UserRepository,
cache: Cache,
logger: Logger,
) -> None:
self._repo = repository
self._cache = cache
self._logger = logger
async def get_user(self, user_id: str) -> User:
# Check cache first
cached = await self._cache.get(f"user:{user_id}")
if cached:
self._logger.info("Cache hit", user_id=user_id)
return User.from_json(cached)
# Fetch from database
user = await self._repo.get_by_id(user_id)
if user:
await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300)
return user
# Production
service = UserService(
repository=PostgresUserRepository(db),
cache=RedisCache(redis),
logger=StructlogLogger(),
)
# Testing
service = UserService(
repository=InMemoryUserRepository(),
cache=FakeCache(),
logger=NullLogger(),
)
Don't expose internal types:
# BAD: Leaking ORM model to API
@app.get("/users/{id}")
def get_user(id: str) -> UserModel: # SQLAlchemy model
return db.query(UserModel).get(id)
# GOOD: Use response schemas
@app.get("/users/{id}")
def get_user(id: str) -> UserResponse:
user = db.query(UserModel).get(id)
return UserResponse.from_orm(user)
Don't mix I/O with business logic:
# BAD: SQL embedded in business logic
def calculate_discount(user_id: str) -> float:
user = db.query("SELECT * FROM users WHERE id = ?", user_id)
orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
# Business logic mixed with data access
# GOOD: Repository pattern
def calculate_discount(user: User, order_history: list[Order]) -> float:
# Pure business logic, easily testable
if len(order_history) > 10:
return 0.15
return 0.0
Weekly Installs
5.4K
Repository
GitHub Stars
32.2K
First Seen
Jan 30, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
opencode4.5K
gemini-cli4.5K
codex4.4K
github-copilot4.2K
kimi-cli3.9K
amp3.9K
React 组合模式指南:Vercel 组件架构最佳实践,提升代码可维护性
102,200 周安装