architecture-patterns by secondsky/claude-skills
npx skills add https://github.com/secondsky/claude-skills --skill architecture-patterns掌握经过验证的后端架构模式,包括整洁架构、六边形架构和领域驱动设计,以构建可维护、可测试和可扩展的系统。
层级(依赖关系向内流动):
关键原则:
组件:
优势:
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
战略模式:
战术模式:
app/
├── domain/ # 实体和业务规则
│ ├── entities/
│ ├── value_objects/
│ └── interfaces/ # 抽象接口
├── use_cases/ # 应用程序业务规则
├── adapters/ # 接口实现
│ ├── repositories/
│ ├── controllers/
│ └── gateways/
└── infrastructure/ # 框架和外部关注点
# domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime
@dataclass
class User:
"""核心用户实体 - 无框架依赖。"""
id: str
email: str
name: str
created_at: datetime
is_active: bool = True
def deactivate(self):
"""业务规则:停用用户。"""
self.is_active = False
def can_place_order(self) -> bool:
"""业务规则:活跃用户可以下单。"""
return self.is_active
# domain/interfaces/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
class IUserRepository(ABC):
"""端口:定义契约,无实现。"""
@abstractmethod
async def find_by_id(self, user_id: str) -> Optional[User]:
pass
@abstractmethod
async def save(self, user: User) -> User:
pass
# use_cases/create_user.py
from dataclasses import dataclass
import uuid
from datetime import datetime
@dataclass
class CreateUserRequest:
email: str
name: str
class CreateUserUseCase:
"""用例:编排业务逻辑。"""
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository
async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
# 业务验证
existing = await self.user_repository.find_by_email(request.email)
if existing:
return CreateUserResponse(user=None, success=False, error="邮箱已存在")
# 创建实体
user = User(
id=str(uuid.uuid4()),
email=request.email,
name=request.name,
created_at=datetime.now(),
is_active=True
)
# 持久化
saved_user = await self.user_repository.save(user)
return CreateUserResponse(user=saved_user, success=True)
# adapters/repositories/postgres_user_repository.py
class PostgresUserRepository(IUserRepository):
"""适配器:PostgreSQL 实现。"""
def __init__(self, pool):
self.pool = pool
async def find_by_id(self, user_id: str) -> Optional[User]:
async with self.pool.acquire() as conn:
row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
return self._to_entity(row) if row else None
async def save(self, user: User) -> User:
async with self.pool.acquire() as conn:
await conn.execute(
"""INSERT INTO users (id, email, name, created_at, is_active)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE SET email = $2, name = $3, is_active = $5""",
user.id, user.email, user.name, user.created_at, user.is_active
)
return user
# 核心领域(六边形中心)
class OrderService:
"""领域服务 - 无基础设施依赖。"""
def __init__(
self,
order_repository: OrderRepositoryPort,
payment_gateway: PaymentGatewayPort,
notification_service: NotificationPort
):
self.orders = order_repository
self.payments = payment_gateway
self.notifications = notification_service
async def place_order(self, order: Order) -> OrderResult:
# 业务逻辑
if not order.is_valid():
return OrderResult(success=False, error="订单无效")
# 使用端口(接口)
payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
if not payment.success:
return OrderResult(success=False, error="支付失败")
order.mark_as_paid()
saved_order = await self.orders.save(order)
await self.notifications.send(
to=order.customer_email,
subject="订单确认",
body=f"订单 {order.id} 已确认"
)
return OrderResult(success=True, order=saved_order)
# 适配器
class StripePaymentAdapter(PaymentGatewayPort):
"""主要适配器:连接到 Stripe API。"""
async def charge(self, amount: Money, customer: str) -> PaymentResult:
# 实现
class MockPaymentAdapter(PaymentGatewayPort):
"""测试适配器:无外部依赖。"""
async def charge(self, amount: Money, customer: str) -> PaymentResult:
return PaymentResult(success=True, transaction_id="mock-123")
# 值对象(不可变)
@dataclass(frozen=True)
class Email:
value: str
def __post_init__(self):
if "@" not in self.value:
raise ValueError("无效的邮箱")
@dataclass(frozen=True)
class Money:
amount: int # 分
currency: str
def add(self, other: "Money") -> "Money":
if self.currency != other.currency:
raise ValueError("货币不匹配")
return Money(self.amount + other.amount, self.currency)
# 实体(具有标识)
class Order:
def __init__(self, id: str, customer: Customer):
self.id = id
self.customer = customer
self.items: List[OrderItem] = []
self.status = OrderStatus.PENDING
self._events: List[DomainEvent] = []
def add_item(self, product: Product, quantity: int):
item = OrderItem(product, quantity)
self.items.append(item)
self._events.append(ItemAddedEvent(self.id, item))
def submit(self):
if not self.items:
raise ValueError("无法提交空订单")
self.status = OrderStatus.SUBMITTED
self._events.append(OrderSubmittedEvent(self.id))
# 聚合(一致性边界)
class Customer:
def __init__(self, id: str, email: Email):
self.id = id
self.email = email
self._addresses: List[Address] = []
def add_address(self, address: Address):
if len(self._addresses) >= 5:
raise ValueError("最多允许 5 个地址")
self._addresses.append(address)
每周安装数
92
代码仓库
GitHub 星标数
90
首次出现
Jan 25, 2026
安全审计
安装于
codex78
gemini-cli77
opencode75
github-copilot73
kimi-cli70
amp69
Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design to build maintainable, testable, and scalable systems.
Layers (dependency flows inward):
Key Principles:
Components:
Benefits:
Strategic Patterns:
Tactical Patterns:
app/
├── domain/ # Entities & business rules
│ ├── entities/
│ ├── value_objects/
│ └── interfaces/ # Abstract interfaces
├── use_cases/ # Application business rules
├── adapters/ # Interface implementations
│ ├── repositories/
│ ├── controllers/
│ └── gateways/
└── infrastructure/ # Framework & external concerns
# domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime
@dataclass
class User:
"""Core user entity - no framework dependencies."""
id: str
email: str
name: str
created_at: datetime
is_active: bool = True
def deactivate(self):
"""Business rule: deactivating user."""
self.is_active = False
def can_place_order(self) -> bool:
"""Business rule: active users can order."""
return self.is_active
# domain/interfaces/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
class IUserRepository(ABC):
"""Port: defines contract, no implementation."""
@abstractmethod
async def find_by_id(self, user_id: str) -> Optional[User]:
pass
@abstractmethod
async def save(self, user: User) -> User:
pass
# use_cases/create_user.py
from dataclasses import dataclass
import uuid
from datetime import datetime
@dataclass
class CreateUserRequest:
email: str
name: str
class CreateUserUseCase:
"""Use case: orchestrates business logic."""
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository
async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
# Business validation
existing = await self.user_repository.find_by_email(request.email)
if existing:
return CreateUserResponse(user=None, success=False, error="Email already exists")
# Create entity
user = User(
id=str(uuid.uuid4()),
email=request.email,
name=request.name,
created_at=datetime.now(),
is_active=True
)
# Persist
saved_user = await self.user_repository.save(user)
return CreateUserResponse(user=saved_user, success=True)
# adapters/repositories/postgres_user_repository.py
class PostgresUserRepository(IUserRepository):
"""Adapter: PostgreSQL implementation."""
def __init__(self, pool):
self.pool = pool
async def find_by_id(self, user_id: str) -> Optional[User]:
async with self.pool.acquire() as conn:
row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
return self._to_entity(row) if row else None
async def save(self, user: User) -> User:
async with self.pool.acquire() as conn:
await conn.execute(
"""INSERT INTO users (id, email, name, created_at, is_active)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE SET email = $2, name = $3, is_active = $5""",
user.id, user.email, user.name, user.created_at, user.is_active
)
return user
# Core domain (hexagon center)
class OrderService:
"""Domain service - no infrastructure dependencies."""
def __init__(
self,
order_repository: OrderRepositoryPort,
payment_gateway: PaymentGatewayPort,
notification_service: NotificationPort
):
self.orders = order_repository
self.payments = payment_gateway
self.notifications = notification_service
async def place_order(self, order: Order) -> OrderResult:
# Business logic
if not order.is_valid():
return OrderResult(success=False, error="Invalid order")
# Use ports (interfaces)
payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
if not payment.success:
return OrderResult(success=False, error="Payment failed")
order.mark_as_paid()
saved_order = await self.orders.save(order)
await self.notifications.send(
to=order.customer_email,
subject="Order confirmed",
body=f"Order {order.id} confirmed"
)
return OrderResult(success=True, order=saved_order)
# Adapters
class StripePaymentAdapter(PaymentGatewayPort):
"""Primary adapter: connects to Stripe API."""
async def charge(self, amount: Money, customer: str) -> PaymentResult:
# Implementation
class MockPaymentAdapter(PaymentGatewayPort):
"""Test adapter: no external dependencies."""
async def charge(self, amount: Money, customer: str) -> PaymentResult:
return PaymentResult(success=True, transaction_id="mock-123")
# Value Objects (immutable)
@dataclass(frozen=True)
class Email:
value: str
def __post_init__(self):
if "@" not in self.value:
raise ValueError("Invalid email")
@dataclass(frozen=True)
class Money:
amount: int # cents
currency: str
def add(self, other: "Money") -> "Money":
if self.currency != other.currency:
raise ValueError("Currency mismatch")
return Money(self.amount + other.amount, self.currency)
# Entities (with identity)
class Order:
def __init__(self, id: str, customer: Customer):
self.id = id
self.customer = customer
self.items: List[OrderItem] = []
self.status = OrderStatus.PENDING
self._events: List[DomainEvent] = []
def add_item(self, product: Product, quantity: int):
item = OrderItem(product, quantity)
self.items.append(item)
self._events.append(ItemAddedEvent(self.id, item))
def submit(self):
if not self.items:
raise ValueError("Cannot submit empty order")
self.status = OrderStatus.SUBMITTED
self._events.append(OrderSubmittedEvent(self.id))
# Aggregates (consistency boundary)
class Customer:
def __init__(self, id: str, email: Email):
self.id = id
self.email = email
self._addresses: List[Address] = []
def add_address(self, address: Address):
if len(self._addresses) >= 5:
raise ValueError("Maximum 5 addresses allowed")
self._addresses.append(address)
Weekly Installs
92
Repository
GitHub Stars
90
First Seen
Jan 25, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykWarn
Installed on
codex78
gemini-cli77
opencode75
github-copilot73
kimi-cli70
amp69
站立会议模板:敏捷开发每日站会指南与工具(含远程团队异步模板)
10,500 周安装
Elastic Observability SLO管理指南:创建、监控服务等级目标与SLI类型详解
204 周安装
CSS开发指南:Flexbox、Grid布局、响应式设计与性能优化最佳实践
199 周安装
OpenSpec (OPSX) 指南:基于工件的开发工作流系统,实现变更管理与自动化
202 周安装
React Testing Library 测试最佳实践:用户行为测试、查询优先级与异步处理指南
200 周安装
Web无障碍测试指南:WCAG合规、键盘导航与屏幕阅读器测试最佳实践
202 周安装
化学分析师技能:原子理论、热力学、光谱学、色谱法、合成路线规划与材料表征
200 周安装