Вопрос или проблема
Хотя я использую Python здесь, мой вопрос не связан с конкретным языком. Я опустил многие детали реализации для краткости, в данный момент это почти псевдокод; и я не привязан к какому-то конкретному фреймворку или ORM. Кроме того, поскольку паттерн Репозиторий часто связан с DDD, я использую термины DDD. Я не утверждаю, что полностью понимаю DDD. Мой проект довольно прост, и DDD может быть совершенно неуместным, но это была возможность узнать о нем. И независимо от того, связан ли он с DDD, я все равно, вероятно, закончу с аналогичными слоями: модели, восстановление экземпляров моделей из базы данных и бизнес-логика.
В одном недавнем проекте мне нужно было добавить бизнес- (и концептуальный объектно-ориентированный) вид поверх сырой SQL-схемы (которая не может быть изменена по причинам наследия; она происходит из открытого инструмента, широко развернутого). Этот “бизнес-уровень” также должен реализовать некоторую тривиальную логику, такую как:
Пользователь
не может быть создан, если он уже существуетПользователь
не может быть добавлен в несуществующуюГруппу
Группа
не может быть удалена, если она все еще содержит пользователей- удаление
Пользователя
приведет к каскадному удалению некоторых связанных с пользователемЭлементов
Я наткнулся на паттерн Репозиторий: в моем случае — и по определению — он помог бы эмулировать концептуальный вид, построенный поверх сырой схемы базы данных. Здесь я могу восстанавливать объекты (корневые агрегаты согласно DDD), взятые из базы данных, и могу добавлять или удалять их из коллекции, скрывая фактические механизмы работы с базой данных.
Итак, в конце концов я получил UserRepository
:
# Модель пользователя (корневой агрегат)
class User:
def __init__(self, name: str, groups: List[Group] = [], items: List[Item] = []):
# некоторые инварианты
if not (groups or items):
raise ValueError("Пользователь должен иметь хотя бы один элемент или должен принадлежать хотя бы одной группе")
self.name = name
self.groups = groups
self.items = items
# Репозиторий пользователя
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 ...") # вставить принадлежности группы
for item in user.items:
self.db_conn.execute(f"INSERT INTO useritem ...") # вставить элементы пользователя
def remove(self, username: str):
self.db_conn.execute(f"DELETE FROM usergroup WHERE ...") # удалить принадлежности группы
self.db_conn.execute(f"DELETE FROM useritem WHERE ...") # удалить элементы
С точки зрения базы данных, объект Пользователь
фактически существует через несколько таблиц (скажем, две таблицы, для упрощения). Вот почему я нашел паттерн Репозиторий полезным здесь: я скрываю эти детали в репозитории, который несет ответственность за восстановление объектов, извлеченных из базы данных, и их распFlattenирование каким-то образом при вставке в базу данных.
Что касается логики пользователя? Согласно DDD, она лежит в 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}' не найдена")
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)
Этот сервис использует как репозитории пользователей, так и репозитории групп. Он содержит бизнес-логику, о которой я упоминал в начале. Моя обеспокоенность, и вы, возможно, предвидите это, касается вызванных исключений: будет ли это означать управление потоком на основе исключений на верхнем уровне приложения? Если да, как этого избежать?
У меня есть два приложения, которые в свою очередь используют этот сервис: веб-приложение (REST API) и приложение на Python. Они будут использовать сервис и доверять ему, чтобы бизнес-правила были соблюдены (нет дублирующихся пользователей, нет пользователей, созданных в несуществующих группах и т. д.).
# Веб-приложение, использующее 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, "Указанный пользователь не существует")
@router.post("/users", status_code=201)
def post_user(user: User) -> User:
try:
return user_service.create(user)
except UserAlreadyExistsException:
raise HTTPException(409, "Указанный пользователь уже существует")
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="Указанный пользователь не существует")
Но я в конечном итоге сталкиваюсь с этим потоком управления на основе исключений, который мне не нравится и который известен как антипаттерн.
Посмотрев на этот популярный пример проекта DDD на Java, я увидел ту же проблему: CustomerService
выбрасывает исключение, которое перехватывается и повторно выбрасывается в связанном веб-сервисе. Кто-то действительно задавал вопрос об этом, но ответа не получил.
Альтернативой было бы переписать логику в моих приложениях, но это привело бы к множеству проблем:
-
UserService
должен будет дублировать операцииUserRepository
, чтобы сделать их доступными - логика домена будет утекать в уровень приложения
- логика домена будет дублироваться в каждом приложении
Ответ или решение
Для предотвращения управления потоком на основе исключений в верхнем уровне паттерна Репозиторий, можно рассмотреть несколько стратегий, которые позволят снизить необходимость выбрасывания исключений для обработки бизнес-логики в ваших сервисах.
1. Введение статуса выполнения операции
Одним из наиболее распространенных подходов является использование объекта результата, который будет содержать как статус выполнения операции, так и возможные ошибки. Вместо того, чтобы выбрасывать исключения, ваши методы могут возвращать объект, который сигнализирует об успешности выполнения и содержит информацию о возможных проблемах.
Пример реализации:
from typing import Optional, Union
class Result:
def __init__(self, success: bool, value: Optional[User] = None, error: Optional[str] = None):
self.success = success
self.value = value
self.error = error
class UserService:
# ...
def get(self, username: str) -> Result:
user = self.user_repo.find_one(username)
if not user:
return Result(success=False, error="UserNotFound")
return Result(success=True, value=user)
def create(self, user: User) -> Result:
if self.user_repo.exists(user.name):
return Result(success=False, error="UserAlreadyExists")
for group in user.groups:
if not self.group_repo.exists(group.name):
return Result(success=False, error=f"Group '{group.name}' not found")
self.user_repo.add(user)
return Result(success=True, value=user)
def delete(self, username: str) -> Result:
if not self.user_repo.exists(username):
return Result(success=False, error="UserNotFound")
self.user_repo.remove(username)
return Result(success=True)
2. Упрощение логики в контроллерах
Контроллеры могут затем проверять статус результата и реагировать соответствующим образом, без необходимости обработки исключений:
@router.get("/users/{username}", status_code=200)
def get_user(username: str) -> User:
result = user_service.get(username)
if not result.success:
if result.error == "UserNotFound":
raise HTTPException(404, "User not found")
return result.value
@router.post("/users", status_code=201)
def post_user(user: User) -> User:
result = user_service.create(user)
if not result.success:
if result.error == "UserAlreadyExists":
raise HTTPException(409, "User already exists")
if result.error.startswith("Group"):
raise HTTPException(422, result.error)
return result.value
@router.delete("/users/{username}", status_code=204)
def delete_user(username: str):
result = user_service.delete(username)
if not result.success and result.error == "UserNotFound":
raise HTTPException(404, "User not found")
3. Использование предикатов для валидации
Ещё одним подходом является использование специальных предикатов для проверки условий ещё до попытки выполнения операции. В этом случае, бизнес-логика будет заключаться в валидирующих методах:
class UserService:
# ...
def validate_user(self, user: User) -> Union[None, str]:
if self.user_repo.exists(user.name):
return "UserAlreadyExists"
for group in user.groups:
if not self.group_repo.exists(group.name):
return f"Group '{group.name}' not found"
return None
def create(self, user: User) -> Result:
validation_error = self.validate_user(user)
if validation_error:
return Result(success=False, error=validation_error)
self.user_repo.add(user)
return Result(success=True, value=user)
Заключение
Используя подходы, описанные выше, вы сможете избавиться от управления потоком, основанного на исключениях, и выделить бизнес-логику более четким образом. Это поможет улучшить читаемость и поддерживаемость кода, а также уменьшить связанность вашего приложения. Вместо исключений вы сможете легко обрабатывать результаты операций, что упрощает тестирование и развитие системы в целом.