處理重復的、復雜的例外處理的最佳方式？

我想重構一些代碼，使其更加簡潔，但我在一個方面有點犯難。我很有可能錯過了一些明顯的東西。

該應用程式是一個 Web 服務器，其中每個 API 端點都是 RequestHandler 類（特別是 Tornado，但我想知道一般的方法）的一個單獨方法。因此，從本質上講，類似于：

的東西

class MyHandler（tornado.web.RequestHandler）。
    def endpoint_1（self, user_id）。
        ...
 
    def endpoint_2(self, user_id) 。
        ...

許多端點需要通過資料庫查找用戶，所以我在類中添加了一個類似def get_user(self, user_id)的方法。該方法還負責對用戶進行各種權限檢查，因此它有幾種失敗的方式，每次失敗都應向用戶發送適當的訊息并關閉連接。

那么我的問題是如何在每個端點方法中以簡明和明顯的方式呼叫該方法。

我認為有三種可能的方法：

1. 在每個端點方法中捕獲所有相關的例外：
。

def endpoint_1（self, user_id）。
   try:
       self.get_user(user_id)
   except UserDoesNotExist。
       self.write("No such user")
       self.connection.close()
       回傳。
   except MissingPermission1。
       self.write("No permission 1")
       self.connection.close()
       回傳。
   ...

這對我來說是最清楚的控制流程，但需要確保這些except:塊被正確地復制到每個方法中。

2. 解壓縮的get_user()呼叫，讓get_user()處理關閉連接
。

def get_user（self, user_id）。
    try:
        user = self.database.lookup(user_id)
        if not user.hasPermission（1）。
            raise MissingPermission1
        ...
    except UserDoesNotExist。
        self.write("No such user")
        self.connection.close()
    except MissingPermission1:
        self.write("沒有權限1")
        self.connection.close()
    ...

def endpoint_1（self, user_id）。
    user = self.get_user(user_id)
    ...

這個選項是迄今為止最簡潔的，但是讓get_user()關閉連接，預先結束endpoint_1()的控制流，我覺得有些 "骯臟"。也許這只是我的問題？檔案可以解決這個問題，但我還是會從中聞到不好的代碼氣味。

3. 除了抓取所有的，傳遞給一個例外處理方法
。

def get_user（self, user_id）。
    user = self.database.lookup(user_id)
    if not user.hasPermission（1）。
        raise MissingPermission1
    ...

def handle_exception（self, exc）。
    if isinstance(ex, MissingPermission1):
        self.write("Missing permission 1")
        回傳。
    ...

def endpoint_1（self, user_id）。
    try:
        self.get_user(user_id)
    except Exception as e:
        self.handle_exception(e)
        self.connection.close()
        return: self.handle_exception(e).

這是我最喜歡的方法，但有些地方還是感覺不對勁。

我很想知道其他人對此的看法，最好是盡可能的概括（例如，不針對龍卷風/webserver）。

uj5u.com熱心網友回復：

我將介紹另一種方法，它在處理桌面GUI時對我很有效。

這個想法類似于你的方法3，但是我們不需要在每個endpoint...方法中添加try/except，而是用我們自己的 "自定義excepthook方法 "覆寫sys.excepthook。這意味著你的endpoint...方法現在可以自由地讓例外一直被引發到自定義excepthook。

這是一個最小的作業例子：

import random
import sys

class UserDoesNotExist（Exception）。pass。
class MissingPermission1（Exception）。pass。

class MyApp:
    def __init__(self):
        sys.excepthook = self.app_excepthook
        
    def app_excepthook（self, exc_type, exc_value, exc_traceback）。
        if issubclass（exc_type, UserDoesNotExist）。
            self.handle_inexistent_user()
        elif issubclass（exc_type, MissingPermission1）。
            self.handle_missing_permission()
        else:
            sys.__excepthook__(exc_type, exc_value, exc_traceback)

    def handle_inexistent_user（self）。
        self.write("No such user")
        # self.connection.close(), etc...。

    def handle_missing_permission（self）。
        self.write("No permission 1")
        # self.connection.close(), etc...。
    
 @staticmethod 
    def write（message）。
        print(message)
    
    def try_to_connect（self）。
        ""模擬一個連接，25%的時間可以作業""。
        r = random.random()
        if r < 0.25:
            raise UserDoesNotExist
        elif r < 0.50:
            raise MissingPermission1
        elif r < 0.75:
            raise ValueError
        else:
            print('Success')

if __name__ == '__main__'/span>:
    app = MyApp()
    app.try_to_connect()

就像我說的，這里的想法是用sys.excepthook方法覆寫app_excepthook。這個方法檢查例外型別并呼叫適當的處理方法。如果沒有方法被委托給某個特定的例外型別，sys.__excepthook__將處理它。這只是一個指向原始 sys.excepthook 的指標，這意味著任何沒有被你的應用程式明確處理的例外都將由 Python 的默認 excepthook 處理。

如果你嘗試執行幾次，你會看到 UserDoesNotExist 和 MissingPermission1 例外被你的 App 正確處理，而 ValueError 則一直到默認的 excepthook。

更重要的是：try_to_connect方法不必擔心任何try/except子句，或者根據例外的型別呼叫一個特定的方法。它只需引發例外并讓應用程式本身來處理它。

您甚至可以進一步抽象化，并將所有的例外處理功能轉移到一個專門的 ExceptionHandler 類中，該類在初始化時成為 MyApp 的實體屬性。

這種方法的一些考慮因素/缺點：

from contextlib import contextmanager


class MyHandler（）。
    def __init__(self, database):
        self.database = database

@contextmanager def get_user(self, user_id) 。
        user = None。
        try:
            user = self.database[user_id]
        except KeyError:
            print(f "缺少用戶{user_id}")

        try:
            yield user
        finally:
            print(f "Closing resources for {user_id}"/span>)
            self.connection = "closed"/span>

    def endpoint_1（self, user_id）。
        self.connection = "open"。
        with self.get_user(user_id) as user:
            if user is None:
                # 處理用戶未被成功檢索的情況。
                print(f "不能處理{user_id}")
            else:
                # Handle scenario if user was successfully retrieved.
                print(f "Successful retrieved {user}")


handler = MyHandler({"a1": "90s Band", "b2": "Banana", "c3": "炸彈"})
print("===== Call 1 ====="/span>)
handler.endpoint_1("a1")
print("===== Call 2 ====="/span>)
handler.endpoint_1("b1")
print("===== Call 3 =====")
handler.endpoint_1("b2")
print("===== Call 4 =====")
handler.endpoint_1("c3")

===== Call 1 =====
成功地檢索到90s Band
關閉資源 for a1
===== 呼叫2 =====
缺少用戶b1
不能處理b1
關閉b1的資源
===== 呼叫3 =====
成功檢索到Banana
關閉b2的資源
===== 呼叫4 =====
成功檢索到炸彈
關閉c3的資源