Одним из недостатков гибких языков, таких как Python, является предположение, что если что-то работает, то скорее всего оно сделано правильно. Я хочу написать скромное руководство по эффективному использованию исключений в Python, правильной их обработке и логировании.
Эффективная обработка исключений
Введение
Давайте рассмотрим следующую систему. У нас есть микросервис, который отвечает за:
· Прослушивание событий о новом заказе;
· Получение заказа из базы данных;
· Проверку состояния принтера;
· Печать квитанции;
· Отправка квитанции в налоговую систему (IRS).
В любой момент может сломаться что угодно. У вас могут возникнуть проблемы с объектом заказа, в котором может не быть нужной информации, или в принтере может закончиться бумага, или же сервис налоговой не будет работать, и вы не сможете синхронизировать с ними квитанцию об оплате, а может быть ваша база данных окажется недоступна.
Ваша задача правильно и проактивно реагировать на любую ситуацию, чтобы избежать ошибок при обработке новых заказов.
И примерно вот такой код на этот случай пишут люди (он, конечно, работает, но плохо и неэффективно):
class OrderService:
def emit(self, order_id: str) -> dict:
try:
order_status = status_service.get_order_status(order_id)
except Exception as e:
logger.exception(
f"Order {order_id} was not found in db "
f"to emit. Error: {e}."
)
raise e
(
is_order_locked_in_emission,
seconds_in_emission,
) = status_service.is_order_locked_in_emission(order_id)
if is_order_locked_in_emission:
logger.info(
"Redoing emission because "
"it was locked in that state after a long time! "
f"Time spent in that state: {seconds_in_emission} seconds "
f"Order: {order_id}, "
f"order_status: {order_status.value}"
)
elif order_status == OrderStatus.EMISSION_IN_PROGRESS:
logger.info("Aborting emission request because it is already in progress!")
return {"order_id": order_id, "order_status": order_status.value}
elif order_status == OrderStatus.EMISSION_SUCCESSFUL:
logger.info(
"Aborting emission because it already happened! "
f"Order: {order_id}, "
f"order_status: {order_status.value}"
)
return {"order_id": order_id, "order_status": order_status.value}
try:
receipt_note = receipt_service.create(order_id)
except Exception as e:
logger.exception(
"Error found during emission! "
f"Order: {order_id}, "
f"exception: {e}"
)
raise e
try:
broker.emit_receipt_note(receipt_note)
except Exception as e:
logger.exception(
"Emission failed! "
f"Order: {order_id}, "
f"exception: {e}"
)
raise e
order_status = status_service.get_order_status(order_id)
return {"order_id": order_id, "order_status": order_status.value}Сначала я сосредоточусь на том, что OrderService слишком много знает, и все эти данные делают его чем-то вроде blob, а чуть позже расскажу о правильной обработке и правильном логировании исключений.
Почему этот сервис — blob?
Этот сервис знает слишком много. Кто-то может сказать, что он знает только то, что ему нужно (то есть все шаги, связанные с формированием чека), но на самом деле он знает куда больше.
Он сосредоточен на создании ошибок (например, база данных, печать, статус заказа), а не на том, что он делает (например, извлекает, проверяет статус, генерирует, отправляет) и на том, как следует реагировать в случае сбоев.
В этом смысле мне кажется, что клиент учит сервис тому, какие исключения он может выдать. Если мы решим переиспользовать его на любом другом этапе (например, клиент захочет получить еще одну печатную копию более старого чека по заказу), мы скопируем большую часть этого кода.
Несмотря на то, что сервис работает нормально, поддерживать его трудно, и неясно, как один шаг соотносится с другим из-за повторяющихся блоков except между шагами, которые отвлекают наше внимание на вопрос «как» вместо того, чтобы думать о «когда».
Первое улучшение: делайте исключения конкретными
Давайте сначала сделаем исключения более точными и конкретными. Преимущества не видны сразу, поэтому я не буду тратить слишком много времени на объяснение этого прямо сейчас. Однако обратите внимание на то, как изменяется код.
Я выделю только то, что мы поменяли:
try:
order_status = status_service.get_order_status(order_id)
except Exception as e:
logger.exception(...)
raise OrderNotFound(order_id) from e
...
try:
...
except Exception as e:
logger.exception(...)
raise ReceiptGenerationFailed(order_id) from e
try:
broker.emit_receipt_note(receipt_note)
except Exception as e:
logger.exception(...)
raise ReceiptEmissionFailed(order_id) from eОбратите внимание, что на этот раз я пользуюсь from e, что является правильным способом создания одного исключения из другого и сохраняет полную трассировку стека.
Второе улучшение: не лезьте не в свое дело
Теперь, когда у нас есть кастомные исключения, мы можем перейти к мысли «не учите классы тому, что может пойти не так» — они сами скажут, если это случится!
# Services
class StatusService:
def get_order_status(order_id):
try:
...
except Exception as e:
raise OrderNotFound(order_id) from e
class ReceiptService:
def create(order_id):
try:
...
except Exception as e:
raise ReceiptGenerationFailed(order_id) from e
class Broker:
def emit_receipt_note(receipt_note):
try:
...
except Exception as e:
raise ReceiptEmissionFailed(order_id) from e
# Main class
class OrderService:
def emit(self, order_id: str) -> dict:
try:
order_status = status_service.get_order_status(order_id)
(
is_order_locked_in_emission,
seconds_in_emission,
) = status_service.is_order_locked_in_emission(order_id)
if is_order_locked_in_emission:
logger.info(
"Redoing emission because "
"it was locked in that state after a long time! "
f"Time spent in that state: {seconds_in_emission} seconds "
f"Order: {order_id}, "
f"order_status: {order_status.value}"
)
elif order_status == OrderStatus.EMISSION_IN_PROGRESS:
logger.info("Aborting emission request because it is already in progress!")
return {"order_id": order_id, "order_status": order_status.value}
elif order_status == OrderStatus.EMISSION_SUCCESSFUL:
logger.info(
"Aborting emission because it already happened! "
f"Order: {order_id}, "
f"order_status: {order_status.value}"
)
return {"order_id": order_id, "order_status": order_status.value}
receipt_note = receipt_service.create(order_id)
broker.emit_receipt_note(receipt_note)
order_status = status_service.get_order_status(order_id)
except OrderNotFound as e:
logger.exception(
f"Order {order_id} was not found in db "
f"to emit. Error: {e}."
)
raise
except ReceiptGenerationFailed as e:
logger.exception(
"Error found during emission! "
f"Order: {order_id}, "
f"exception: {e}"
)
raise
except ReceiptEmissionFailed as e:
logger.exception(
"Emission failed! "
f"Order: {order_id}, "
f"exception: {e}"
)
raise
else:
return {"order_id": order_id, "order_status": order_status.value}Как вам? Намного лучше, правда? У нас есть один блок try, который построен достаточно логично, чтобы понять, что произойдет дальше. Вы сгруппировали конкретные блоки, за исключением тех, которые помогают вам понять «когда» и крайние случаи. И, наконец, у вас есть блок else, в котором описано, что произойдет, если все отработает как надо.
Кроме того, пожалуйста, обратите внимание на то, что я сохранил инструкции raise без объявления объекта исключения. Это не опечатка. На самом деле, это правильный способ повторного вызова исключения: простой и немногословный.
Но это еще не все. Логирование продолжает меня раздражать.
Третье улучшение: улучшение логирования
Этот шаг напоминает мне принцип «говори, а не спрашивай», хотя это все же не совсем он. Вместо того, чтобы запрашивать подробности исключения и на их основе выдавать полезные сообщения, исключения должны выдавать их сами – в конце концов, я их конкретизировал!
### Exceptions
class OrderCreationException(Exception):
pass
class OrderNotFound(OrderCreationException):
def __init__(self, order_id):
self.order_id = order_id
super().__init__(
f"Order {order_id} was not found in db "
"to emit."
)
class ReceiptGenerationFailed(OrderCreationException):
def __init__(self, order_id):
self.order_id = order_id
super().__init__(
"Error found during emission! "
f"Order: {order_id}"
)
class ReceiptEmissionFailed(OrderCreationException):
def __init__(self, order_id):
self.order_id = order_id
super().__init__(
"Emission failed! "
f"Order: {order_id} "
)
### Main class
class OrderService:
def emit(self, order_id: str) -> dict:
try:
...
except OrderNotFound:
logger.exception("We got a database exception")
raise
except ReceiptGenerationFailed:
logger.exception("We got a problem generating the receipt")
raise
except ReceiptEmissionFailed:
logger.exception("Unable to emit the receipt")
raise
else:
return {"order_id": order_id, "order_status": order_status.value}Наконец-то мои глаза чувствуют облегчение. Поменьше повторений, пожалуйста! Примите к сведению, что рекомендуемый способ выглядит так, как я написал его выше: logger.exception("ЛЮБОЕ СООБЩЕНИЕ"). Вам даже не нужно передавать исключение, поскольку его наличие уже подразумевается. Кроме того, кастомное сообщение, которое мы определили в каждом исключении с идентификатором order_id, будет отображаться в логах, поэтому вам не нужно повторяться и не нужно оперировать внутренними данными об исключениях.
Вот пример вывода ваших логов:
❯ python3 testme.py
Unable to emit the receipt # <<-- My log message
Traceback (most recent call last):
File "/path/testme.py", line 19, in <module>
tryme()
File "/path/testme.py", line 14, in tryme
raise ReceiptEmissionFailed(order_id)
ReceiptEmissionFailed: Emission failed! Order: 10 # <<-- My exception messageТеперь всякий раз, когда я получаю это исключение, сообщение уже ясно и понятно, и мне не нужно помнить о логировании order_id, который я сгенерировал.
Последнее улучшение: упрощение
После более детального рассмотрения нашего окончательного кода, он кажется лучше, теперь его легко читать и поддерживать.
Но управляет ли OrderService бизнес-логикой? Я не думаю, что это сервис в общем смысле. Он больше похож на координацию вызовов настоящих сервисов обеспечивающих бизнес-логику, которая выглядит получше, чем паттерн facade.
Кроме того, можно заметить, что он запрашивает данные у status_service, чтобы что-то с ними сделать. (Что, на этот раз, действительно разрушает идею «Говори, а не спрашивай»).
Перейдем к упрощению.
class OrderFacade: # Renamed to match what it actually is
def emit(self, order_id: str) -> dict:
try:
# NOTE: info logging still happens inside
status_service.ensure_order_unlocked(order_id)
receipt_note = receipt_service.create(order_id)
broker.emit_receipt_note(receipt_note)
order_status = status_service.get_order_status(order_id)
except OrderAlreadyInProgress as e:
# New block
logger.info("Aborting emission request because it is already in progress!")
return {"order_id": order_id, "order_status": e.order_status.value}
except OrderAlreadyEmitted as e:
# New block
logger.info(f"Aborting emission because it already happened! {e}")
return {"order_id": order_id, "order_status": e.order_status.value}
except OrderNotFound:
logger.exception("We got a database exception")
raise
except ReceiptGenerationFailed:
logger.exception("We got a problem generating the receipt")
raise
except ReceiptEmissionFailed:
logger.exception("Unable to emit the receipt")
raise
else:
return {"order_id": order_id, "order_status": order_status.value}Мы только что создали новый метод ensure_order_unlocked для нашего status_service, который теперь отвечает за создание исключений/логирование в случае, если что-то идет не так.
Хорошо, а теперь скажите, насколько легче теперь стало это читать?
Я могу понять все return при беглом просмотре. Я знаю, что происходит, когда все идет хорошо, и как крайние случаи могут привести к разным результатам. И все это без прокрутки взад-вперед.
Теперь этот код такой же простой, каким (в основном) должен быть любой код.
Обратите внимание, что я решил вывести объект исключения e в логах, поскольку под капотом он будет запускать str(e), который вернет сообщение об исключении. Я подумал, что было бы полезно говорить подробно, поскольку мы не используем log.exception для этого блока, поэтому сообщение об исключении не будет отображаться.
Теперь давайте разберемся с некоторыми хитростями, которые помогут вам сделать код понятным для чтения и простым в обслуживании.
Эффективное создание исключений
Всегда классифицируйте свои исключения через базовое и расширяйте все конкретные исключения от него. С помощью этой полезной практики вы можете переиспользовать логику для связанного кода.
Исключения – это объекты, которые несут в себе информацию, поэтому не стесняйтесь добавлять кастомные атрибуты, которые могут помочь вам понять, что происходит. Не позволяйте своему бизнес-коду учить вас тому, как он должен быть построен, ведь с таким количеством сообщений и деталей потерять себя становится трудно.
# Base category exception
class OrderCreationException(Exception):
pass
# Specific error with custom message. Order id is required.
class OrderNotFound(OrderCreationException):
def __init__(self, order_id):
self.order_id = order_id # custom property
super().__init__(
f"Order {order_id} was not found in db "
f"to emit."
)
# Specific error with custom message. Order id is required.
class ReceiptGenerationFailed(OrderCreationException):
def __init__(self, order_id):
self.order_id = order_id # custom property
super().__init__(
"Error found during emission! "
f"Order: {order_id}"
)В примере выше я мог бы выйти за рамки и расширить базовый класс, чтобы всегда получать order_id, если мне это нужно. Этот совет поможет сохранить код сухим, поскольку мне не нужно быть многословным при создании исключений. Так можно использовать всего лишь одну переменную.
def func1(order_id):
raise OrderNotFound(order_id)
# instead of raise OrderNotFound(f"Can't find order {order_id}")
def func2(order_id):
raise OrderNotFound(order_id)
# instead of raise OrderNotFound(f"Can't find order {order_id}") В тестировании также будет больше смысла, поскольку я могу сделать assert order_id через строку.
assert e.order_id == order_id
# instead of assert order_id in str(e)Ловим и создаем исключения эффективно
Еще одна вещь, которую люди часто делают неправильно – это отлавливают и повторно создают исключения.
Согласно PEP 3134 Python, делать нужно следующим образом.
Повторное создание исключения
Обычной инструкции raise более чем достаточно.
try:
...
except CustomException as ex:
# do stuff (e.g. logging)
raiseСоздание одного исключения из другого
Этот вариант особо актуален, поскольку он сохраняет всю трассировку стека и помогает вашей команде отлаживать основные проблемы.
try:
...
except CustomException as ex:
raise MyNewException() from exЭффективное логирование исключений
Еще один совет, который не позволит вам быть слишком многословным.
Используйте logger.exception
Вам не нужно логировать объект исключения. Функция exception логгера предназначена для использования внутри блоков except. Она уже обрабатывает трассировку стека с информацией о выполнении и отображает, какое исключение вызвало ее, с сообщением, установленном на уровне ошибки!
try:
...
except CustomException:
logger.exception("custom message")А что, если это не ошибка?
Если по какой-то причине вы не хотите логировать исключение как ошибку, то возможно, это предупреждение или просто информация, как было показано выше.
Вы можете принять решение установить exc_info в True, если хотите сохранить трассировку стека. Кроме того, было бы неплохо использовать объект исключения внутри сообщения.
Источники
Документация Python:
· Python logging.logger.exception
· Python PEP 3134
Принципы и качество кода:
· Blob
Материал подготовлен в рамках курса "Python Developer. Professional".
Всех желающих приглашаем на онлайн-интенсив «Быстрая разработка JSON API приложений на Flask». На этом интенсиве мы:
— Познакомимся со спецификацией JSON API;
— Узнаем, что такое сериализация/десериализация данных;
— Узнаем, что такое marshmallow и marshmallow-jsonapi;
— Познакомимся со Swagger;
— Посмотрим на обработку и выдачу связей.