如何避免Repository模式上层应用的异常式控制流？

本文以Python为例，但问题不局限于特定语言，代码已简化为类伪代码，不绑定特定框架或ORM。因Repository模式常与DDD关联，故使用DDD术语，项目简单可能不适用DDD，但仍会有模型、数据重构、业务逻辑等分层。

项目背景 #

近期项目需基于无法修改的遗留SQL架构（来自广泛部署的开源工具）构建面向业务的面向对象视图，业务层需实现以下规则：

• 若User已存在则无法创建
• 无法将User添加至不存在的Group
• 若Group仍包含用户则无法删除
• 删除User时需级联删除关联的Item

UserRepository实现 #

采用Repository模式封装数据访问，隐藏数据库细节，实现聚合根的对象重构与持久化，以下是UserRepository的实现：

# User model (aggregate root)
class User:
  def __init__(self, name: str, groups: List[Group] = [], items: List[Item] = []):
    # some invariant
    if not (groups or items):
      raise ValueError("A user must have at least one item or must belong to at least one group")
    self.name = name
    self.groups = groups
    self.items = items

# User repository
class UserRepository:
  def __init__(self, db_conn):
    self.db_conn = db_conn
  
  def exists(self, username: str) -> bool: # ...
  def find_all(self) -> list[User]: # ...
  def find_one(self, username: str) -> User:
    if not self.exists(username):
      return None
    groups = self.db_conn.execute(f"SELECT FROM usergroup WHERE ...")
    items = self.db_conn.execute(f"SELECT FROM useritem WHERE ...")
    return User(username, groups, items)
  
  def add(self, user: User):
    for group in user.groups:
      self.db_conn.execute(f"INSERT INTO usergroup ...") # insert group belongings
    for item in user.items:
      self.db_conn.execute(f"INSERT INTO useritem ...") # insert user items
    
  def remove(self, username: str):
    self.db_conn.execute(f"DELETE FROM usergroup WHERE ...") # delete group belongings
    self.db_conn.execute(f"DELETE FROM useritem WHERE ...") # delete items

从数据库视角，User对象分散在多张表中，Repository模式可隐藏这些细节，负责对象重构与持久化。

UserService业务逻辑实现 #

业务逻辑在UserService中实现：

class UserAlreadyExistsException(Exception): # ...
class UserNotFoundException(Exception): # ...
class GroupNotFoundException(Exception): # ...

class UserService:
  def __init__(self, user_repo: UserRepository, group_repo: GroupRepository):
    self.user_repo = user_repo
    self.group_repo = group_repo
  
  def get(self, username: str) -> User:
    user = self.user_repo.find_one(username)
    if not user:
      raise UserNotFoundException
    return user

  def create(self, user: User) -> User:
    if self.user_repo.exists(user.name):
      raise UserAlreadyExistsException
      
    for group in user.groups:
      if not self.group_repo.exists(group.name):
        raise GroupNotFoundException(f"'{group.name}' has not been found")
    
    self.user_repo.add(user)
    return user
  
  def delete(self, username: str):
    if not self.user_repo.exists(username):
      raise UserNotFoundException
    self.user_repo.remove(username)

异常式控制流的问题 #

该服务包含业务规则，但抛出的异常导致上层Web应用（REST API）和Python应用不得不采用异常式控制流，这被视为反模式：

# Web app consuming UserService

@router.get("/users/{username}", status_code=200)
def get_user(username: str) -> User:
  try:
    return user_service.get(username)
  except UserNotFoundException:
    raise HTTPException(404, "Given user does not exist")

@router.post("/users", status_code=201)
def post_user(user: User) -> User:
  try:
    return user_service.create(user)
  except UserAlreadyExistsException:
    raise HTTPException(409, "Given user already exists")
  except GroupNotFoundException as exc:
    raise HTTPException(422, str(exc))

@router.delete("/users/{username}", status_code=204)
def delete_user(username: str):
  try:
    user_service.delete(username)
  except UserNotFoundException:
    raise HTTPException(404, detail="Given user does not exist")

若将业务逻辑移至应用层，会导致UserService需重复Repository操作、领域逻辑泄漏至应用层、逻辑在多应用中重复等问题。

核心问题 #

如何避免Repository模式上层应用采用异常式控制流？

补充说明 #

• 2019年有文章阐述过该问题并给出替代方案
• 2023年有文章的案例也存在相同的异常式控制流问题：

class PaymentController {

    public Response handlePayment(Request request, int amountInCents) {
        try {
          this.paymentService.createPaymentForUser(request.getUser(), amountInCents);
          return new Response(200, new HashMap<>());
        } catch (PaymentValidationException ex) {
          throw new ResponseStatusException(400, ex.getMessage())
        } catch (UserNotFoundException ex) {
          throw new ResponseStatusException(401, "You are not authorized to create payments for this user");
        }
    }
}

该案例与本问题高度相似，但未提及解决方法或替代方案。

解决方案 #

1. 结果对象模式（Result Pattern） #

定义通用的结果类，包含成功状态、返回数据、错误类型和信息，让Service方法返回该对象而非抛出异常，从根源上避免异常式控制流。

定义Result类

from typing import Generic, TypeVar, Optional

T = TypeVar('T')

class Result(Generic[T]):
    def __init__(self, success: bool, data: Optional[T] = None, error_type: Optional[str] = None, error_message: Optional[str] = None):
        self.success = success
        self.data = data
        self.error_type = error_type
        self.error_message = error_message

    @staticmethod
    def success(data: T) -> 'Result[T]':
        return Result(True, data=data)

    @staticmethod
    def failure(error_type: str, error_message: str) -> 'Result[T]':
        return Result(False, error_type=error_type, error_message=error_message)

修改UserService方法

class UserService:
    def __init__(self, user_repo: UserRepository, group_repo: GroupRepository):
        self.user_repo = user_repo
        self.group_repo = group_repo
    
    def get(self, username: str) -> Result[User]:
        user = self.user_repo.find_one(username)
        if not user:
            return Result.failure("USER_NOT_FOUND", "Given user does not exist")
        return Result.success(user)

    def create(self, user: User) -> Result[User]:
        if self.user_repo.exists(user.name):
            return Result.failure("USER_ALREADY_EXISTS", "Given user already exists")
        
        for group in user.groups:
            if not self.group_repo.exists(group.name):
                return Result.failure("GROUP_NOT_FOUND", f"'{group.name}' has not been found")
        
        self.user_repo.add(user)
        return Result.success(user)
    
    def delete(self, username: str) -> Result[None]:
        if not self.user_repo.exists(username):
            return Result.failure("USER_NOT_FOUND", "Given user does not exist")
        self.user_repo.remove(username)
        return Result.success(None)

修改Web层代码

无需try-catch，直接根据结果状态处理：

@router.get("/users/{username}", status_code=200)
def get_user(username: str):
    result = user_service.get(username)
    if not result.success:
        raise HTTPException(404, result.error_message)
    return result.data

@router.post("/users", status_code=201)
def post_user(user: User):
    result = user_service.create(user)
    if not result.success:
        if result.error_type == "USER_ALREADY_EXISTS":
            raise HTTPException(409, result.error_message)
        elif result.error_type == "GROUP_NOT_FOUND":
            raise HTTPException(422, result.error_message)
    return result.data

@router.delete("/users/{username}", status_code=204)
def delete_user(username: str):
    result = user_service.delete(username)
    if not result.success:
        raise HTTPException(404, result.error_message)

2. 轻量方案：返回元组/可选类型 #

如果不想定义通用Result类，可让Service方法返回(数据, 错误标识)的元组，或结合Optional类型使用：

# 修改UserService的get方法
def get(self, username: str) -> tuple[Optional[User], Optional[str]]:
    user = self.user_repo.find_one(username)
    if not user:
        return None, "USER_NOT_FOUND"
    return user, None

# Web层处理
@router.get("/users/{username}", status_code=200)
def get_user(username: str):
    user, error = user_service.get(username)
    if error == "USER_NOT_FOUND":
        raise HTTPException(404, "Given user does not exist")
    return user

这种方式更轻量，但扩展性不如Result模式，适合简单业务场景。

3. 增强类型安全：领域错误枚举 #

将错误类型定义为枚举，替代字符串标识，避免硬编码错误：

from enum import Enum

class DomainError(Enum):
    USER_NOT_FOUND = "USER_NOT_FOUND"
    USER_ALREADY_EXISTS = "USER_ALREADY_EXISTS"
    GROUP_NOT_FOUND = "GROUP_NOT_FOUND"

# 更新Result类的error_type为枚举类型
class Result(Generic[T]):
    def __init__(self, success: bool, data: Optional[T] = None, error: Optional[DomainError] = None, error_message: Optional[str] = None):
        self.success = success
        self.data = data
        self.error = error
        self.error_message = error_message

    @staticmethod
    def failure(error: DomainError, error_message: str) -> 'Result[T]':
        return Result(False, error=error, error_message=error_message)

Web层判断错误时可直接使用枚举值，提升代码的严谨性。

方案优势 #

• 避免异常式控制流，代码逻辑更直观，符合"正常路径优先"的阅读习惯
• 错误信息和类型明确，上层应用可直接根据结果分支处理
• 领域逻辑仍保留在Service层，不会泄漏到应用层
• 扩展性强，新增错误类型只需扩展Result类或枚举即可

内容的提问来源于stack exchange，提问作者insideshell