References: Martin Fowler, Patterns of Enterprise Application Architecture, Repository / Unit of Work; official documentation of each ORM; High Performance MySQL.
Applicable Scenarios
- ORM / Repository implementation
- Cross-ORM common patterns
- N+1 query prevention
- Transaction management (declarative / programmatic)
- Complex queries / custom SQL
- Read-write splitting / multiple data sources
Core Principles
1. Repository 隐藏 ORM 细节
Service 层不知道是 JPA 还是 MyBatis 还是 Prisma
-
Return Domain Model, not ORM entity (if different) Or: DB layer = Domain layer (unified)
-
N+1 prevention is a default requirement Proactively use join fetch / eager load / select_related
-
Transaction boundary = business operation boundary Not in Controller, not across multiple aggregates
-
Do not make external calls inside a transaction HTTP / email / third party → long transaction → deadlock
-
Use SQL / Native Query for complex queries ORM is not a universal hammer
-
Treat write performance vs read performance differently Write: fewer indexes, bulk insert Read: more indexes, caching
## Repository 模式
### Java + Spring Data JPA
```java
public interface OrderRepository extends JpaRepository<Order, Long> {
// 自动派生
List<Order> findByUserIdAndStatus(Long userId, OrderStatus status);
// 自定义 JPQL（防 N+1）
@Query("SELECT o FROM Order o " +
"JOIN FETCH o.items i " +
"JOIN FETCH i.product " +
"WHERE o.userId = :userId")
List<Order> findByUserIdWithItems(@Param("userId") Long userId);
// 原生 SQL
@Query(value = "SELECT * FROM orders WHERE created_at > :since", nativeQuery = true)
List<Order> findRecent(@Param("since") LocalDateTime since);
// 修改
@Modifying
@Query("UPDATE Order o SET o.status = :status WHERE o.id IN :ids")
int updateStatus(@Param("ids") List<Long> ids, @Param("status") OrderStatus status);
}
// Service 用
@Service
@Transactional
public class OrderService {
public List<OrderDto> getUserOrders(Long userId) {
// 用 join fetch 防 N+1
return orderRepository.findByUserIdWithItems(userId).stream()
.map(OrderDto::from)
.toList();
}
}
TypeScript + Prisma
// schema.prisma
model Order {
id Int @id @default(autoincrement())
userId Int
status OrderStatus
items OrderItem[]
createdAt DateTime @default(now())
user User @relation(fields: [userId], references: [id])
}
// Repository
@Injectable()
export class OrderRepository {
constructor(private prisma: PrismaService) {}
// 防 N+1：用 include
async findByUserId(userId: number) {
return this.prisma.order.findMany({
where: { userId },
include: {
items: {
include: { product: true }
}
}
});
}
// 复杂查询：用 raw query
async findRecent(days: number) {
return this.prisma.$queryRaw<Order[]>`
SELECT * FROM "Order"
WHERE created_at > NOW() - INTERVAL '${days} days'
`;
}
// 事务
async transferOrder(orderId: number, newUserId: number) {
return this.prisma.$transaction(async (tx) => {
const order = await tx.order.findUnique({ where: { id: orderId } });
if (!order) throw new NotFoundException();
return tx.order.update({
where: { id: orderId },
data: { userId: newUserId, transferredAt: new Date() }
});
});
}
}
TypeScript + TypeORM
@EntityRepository(Order)
export class OrderRepository extends Repository<Order> {
async findByUserIdWithItems(userId: number): Promise<Order[]> {
return this.createQueryBuilder('order')
.leftJoinAndSelect('order.items', 'item')
.leftJoinAndSelect('item.product', 'product')
.where('order.userId = :userId', { userId })
.getMany();
}
}
Python + Django ORM
class OrderRepository:
@staticmethod
def find_by_user_id(user_id):
# 防 N+1：select_related (一对一/一对多反向) + prefetch_related (一对多/多对多)
return Order.objects.filter(user_id=user_id) \
.select_related('user') \
.prefetch_related('items__product') \
.order_by('-created_at')
@staticmethod
def find_recent(days):
return Order.objects.filter(
created_at__gt=timezone.now() - timedelta(days=days)
)
@staticmethod
@transaction.atomic
def bulk_update_status(order_ids, new_status):
return Order.objects.filter(id__in=order_ids).update(status=new_status)
Python + SQLAlchemy 2
class OrderRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def find_by_user_id(self, user_id: int) -> list[Order]:
# 防 N+1：selectinload / joinedload
stmt = (
select(Order)
.where(Order.user_id == user_id)
.options(
selectinload(Order.items).selectinload(OrderItem.product)
)
.order_by(Order.created_at.desc())
)
result = await self.session.execute(stmt)
return result.scalars().all()
async def find_recent(self, days: int) -> list[Order]:
stmt = select(Order).where(
Order.created_at > datetime.utcnow() - timedelta(days=days)
)
result = await self.session.execute(stmt)
return result.scalars().all()
Go + GORM
type OrderRepository struct {
db *gorm.DB
}
func (r *OrderRepository) FindByUserID(ctx context.Context, userID int64) ([]Order, error) {
var orders []Order
err := r.db.WithContext(ctx).
Preload("Items.Product"). // 防 N+1
Where("user_id = ?", userID).
Order("created_at DESC").
Find(&orders).Error
return orders, err
}
func (r *OrderRepository) Transaction(ctx context.Context, fn func(*gorm.DB) error) error {
return r.db.WithContext(ctx).Transaction(fn)
}
Go + sqlc (recommended: type safety + performance)
-- queries/order.sql
-- name: GetOrderByID :one
SELECT * FROM orders WHERE id = $1;
-- name: ListOrdersByUser :many
SELECT * FROM orders WHERE user_id = $1 ORDER BY created_at DESC;
-- name: UpdateOrderStatus :exec
UPDATE orders SET status = $2 WHERE id = $1;
// 自动生成的代码（类型安全）
queries := order.New(db)
o, err := queries.GetOrderByID(ctx, 123)
orders, err := queries.ListOrdersByUser(ctx, userID)
N+1 Problem (must understand)
Counterexample (most common)
// 1 个查询返回 100 个订单
const orders = await orderRepo.findAll();
// 100 次查询拿 user
for (const order of orders) {
console.log(order.user.name); // ← 触发懒加载，每个发 SQL
}
// 总查询数：1 + 100 = 101
Solutions
// Prisma：include
const orders = await prisma.order.findMany({
include: { user: true }
});
// TypeORM：leftJoinAndSelect
const orders = await this.repo.createQueryBuilder('order')
.leftJoinAndSelect('order.user', 'user')
.getMany();
// JPA：JOIN FETCH
@Query("SELECT o FROM Order o JOIN FETCH o.user WHERE ...")
// Django：select_related
Order.objects.select_related('user').all()
// SQLAlchemy：joinedload
session.query(Order).options(joinedload(Order.user)).all()
// 总查询数：1
Advanced: DataLoader pattern (commonly used in GraphQL)
// 批量收集 + 单次查询
const userLoader = new DataLoader(async (userIds: number[]) => {
const users = await prisma.user.findMany({
where: { id: { in: userIds } }
});
return userIds.map(id => users.find(u => u.id === id));
});
// 用法
const user1 = await userLoader.load(1);
const user2 = await userLoader.load(2);
// 实际只发 1 次 SQL（IN [1, 2]）
Transaction Management
Declarative (recommended)
// Spring：@Transactional
@Service
@Transactional
public class OrderService {
public Order createOrder(...) {
Order order = ...;
orderRepo.save(order);
inventoryRepo.decrement(...); // 同一事务
return order;
}
}
// NestJS + TypeORM 或 Prisma
@Injectable()
export class OrderService {
async createOrder(dto: CreateOrderDto) {
return this.prisma.$transaction(async (tx) => {
const order = await tx.order.create({ data: ... });
await tx.inventory.update({ where: ..., data: { stock: { decrement: 1 } } });
return order;
});
}
}
# Django: @transaction.atomic
@transaction.atomic
def create_order(...):
order = Order.objects.create(...)
Inventory.objects.filter(...).update(stock=F('stock') - 1)
return order
Programmatic
// 需要细粒度控制时
public Order createOrder(...) {
return transactionTemplate.execute(status -> {
Order order = orderRepo.save(...);
inventoryRepo.decrement(...);
return order;
});
}
Isolation Levels
@Transactional(isolation = Isolation.SERIALIZABLE)
public void criticalOperation() { ... }
// 默认 READ_COMMITTED 已能满足大多数业务
// 涉及金融并发用 SERIALIZABLE + 重试
Nested Transactions (use with caution)
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void independent() {
// 独立事务，外层失败不影响
}
Bulk Operations
// JPA
@Modifying
@Query("UPDATE Order o SET o.status = :status WHERE o.id IN :ids")
int batchUpdate(@Param("ids") List<Long> ids, @Param("status") OrderStatus status);
// 批量插入（需配置 hibernate.jdbc.batch_size）
List<Order> orders = ...;
orderRepo.saveAll(orders);
// Prisma
await prisma.order.createMany({
data: [...],
skipDuplicates: true,
});
// Django
Order.objects.bulk_create([...], batch_size=1000);
// 性能差距：循环单条 vs 批量 = 100x+
Read-Write Splitting
// Spring：@Transactional(readOnly = true) 路由到从库
@Transactional(readOnly = true)
public List<Order> getAllOrders() {
return orderRepo.findAll();
}
// Prisma：单独 client
const readReplica = new PrismaClient({
datasources: { db: { url: process.env.DATABASE_REPLICA_URL } }
});
// 读用 readReplica，写用 prisma
Multiple Data Sources
// Spring：@Transactional("secondaryTransactionManager")
@Service
public class CrossDatabaseService {
@Transactional("primaryTransactionManager")
public void writeToPrimary() { ... }
@Transactional("secondaryTransactionManager")
public void writeToSecondary() { ... }
}
Workflow
1. 设计 Repository 接口
- 业务驱动：findByOrderNumber 而非 findById
- 不暴露 ORM 类型
↓
2. 实现 Repository
- 用 ORM 或 SQL 构建器
- 防 N+1（join fetch / include）
↓
3. 事务策略
- 声明式 @Transactional
- 边界 = 业务操作边界
↓
4. 性能优化
- 批量操作
- 读写分离
- 自定义 SQL
↓
5. 测试
- 用真实 DB 或 Testcontainers
- 不用 Mock 数据库
Supporting Templates
templates/repository-template.md— Repository interface / implementation / test templates (multiple ORMs)
Quality Self-Check
□ Repository 接口面向业务，不面向技术
□ 返回 Domain Model（如分层）
□ 默认防 N+1（join fetch / include）
□ 事务边界明确
□ 事务不包含外部调用
□ 批量操作使用 bulk_*
□ 自定义 SQL 有参数化（防注入）
□ 读多用 readOnly 事务（路由从库）
□ 大查询有 limit
□ 测试用真实 DB（Testcontainers）
□ 慢查询监控（pg_stat_statements / 慢日志）
□ ORM 配置 batch_size / fetch_size
Common Pitfalls
- N+1 query——performance collapse, fixed by adding an include
- Transaction wrapping external calls——HTTP 30 seconds causes lock
- Loop INSERT——1000 SQL statements, should use bulk_create
- OFFSET deep pagination——offset 100000 is deathly slow
- Not using parameterized queries——SQL injection
- ORM automatically generates SELECT *——transmission waste
- Lazy loading triggered in Controller——transaction already closed, error
- Using DTO directly as entity——writes all fields on update
- Not limiting query scope——
findAll()large table collapse - Bulk update without batching——table lock
- Using distributed transactions (2PC) across services——use with caution, should use Saga
- Mistaking N+1 for an ORM bug——actually a usage error
- Not monitoring slow SQL——only find out when it blows up in production
- Transaction nesting levels out of control——misunderstanding propagation mechanisms
Collaboration with Other Skills
上游：
domain-modeling → Domain 模型
database 工作流 → 表结构 / 索引
下游：
api-implementation → Service 调 Repository
caching-strategy → 缓存读 vs DB 读
observability → SQL 性能监控
testing-implementation → 用 Testcontainers
Related References
- Project root
references/backend-tech-stack-guide.md— ORM selection - database-engineer workflow query-review skill