錯誤處理的藝術：開發強健的應用程式
在軟體開發過程中，錯誤處理如同房子的地基，穩固的地基才能支撐起整棟建築。一個健壯的應用程式，必須能有效處理各種預期和非預期的錯誤，提供使用者友善的回饋，並確保系統穩定執行。我將分享在錯誤管理方面的一些心得體會，並以 RoomListRequest 的改進為例，說明如何建構更具彈性的錯誤處理機制。
重新思考 RoomListRequest：有效性驗證
Room 模型擁有多個屬性，這導致過濾條件的組合數量非常龐大。為了簡化設計，我做了以下限定：
code屬性僅支援__eq操作，用於查詢特定程式碼的房間。price屬性支援__eq、__lt和__gt操作，用於價格範圍的篩選。- 其他屬性不支援過濾。
請求的有效性驗證應該在到達 Use Case 之前完成，而不是由 Use Case 負責檢查輸入值的格式和有效性。因此，我將 RoomListRequest 拆分為 RoomListValidRequest 和 RoomListInvalidRequest 兩個類別，並建立一個工廠函式 build_room_list_request 來傳回相應的物件。
from collections.abc import Mapping
class RoomListInvalidRequest:
def __init__(self):
self.errors = []
def add_error(self, parameter, message):
self.errors.append({"parameter": parameter, "message": message})
def has_errors(self):
return len(self.errors) > 0
def __bool__(self):
return False
class RoomListValidRequest:
def __init__(self, filters=None):
self.filters = filters
def __bool__(self):
return True
def build_room_list_request(filters=None):
accepted_filters = ["code__eq", "price__eq", "price__lt", "price__gt"]
invalid_req = RoomListInvalidRequest()
if filters is not None:
if not isinstance(filters, Mapping):
invalid_req.add_error("filters", "Is not iterable")
return invalid_req
for key, value in filters.items():
if key not in accepted_filters:
invalid_req.add_error("filters", f"Key {key} cannot be used")
if invalid_req.has_errors():
return invalid_req
return RoomListValidRequest(filters=filters)
內容解密：
這段程式碼定義了 RoomListInvalidRequest 和 RoomListValidRequest 兩個類別，分別代表無效和有效的房間列表請求。build_room_list_request 函式根據傳入的 filters 引數，檢查其型別和鍵值是否有效，並傳回相應的請求物件。
Response 物件的設計：一致的錯誤回饋
Use Case 執行過程中可能出現各種錯誤，例如驗證錯誤、業務邏輯錯誤、資料函式庫錯誤等。無論錯誤型別如何，Use Case 都應該傳回結構一致的 Response 物件，以便客戶端程式碼進行處理。
我設計了 ResponseSuccess 和 ResponseFailure 兩個類別，分別代表成功和失敗的回應。ResponseFailure 物件包含錯誤型別和錯誤訊息，方便客戶端程式碼根據錯誤型別採取不同的處理策略。
class ResponseTypes:
PARAMETERS_ERROR = "ParametersError"
RESOURCE_ERROR = "ResourceError"
SYSTEM_ERROR = "SystemError"
SUCCESS = "Success"
class ResponseFailure:
def __init__(self, type_, message):
self.type = type_
self.message = self._format_message(message)
def _format_message(self, msg):
if isinstance(msg, Exception):
return f"{msg.__class__.__name__}: {msg}"
return msg
@property
def value(self):
return {"type": self.type, "message": self.message}
def __bool__(self):
return False
class ResponseSuccess:
def __init__(self, value=None):
self.type = ResponseTypes.SUCCESS
self.value = value
def __bool__(self):
return True
def build_response_from_invalid_request(invalid_request):
message = "\n".join(
[f"{err['parameter']}: {err['message']}" for err in invalid_request.errors]
)
return ResponseFailure(ResponseTypes.PARAMETERS_ERROR, message)
內容解密：
這段程式碼定義了 ResponseTypes 列舉類別，用於表示不同的回應型別。ResponseFailure 類別用於表示失敗的回應，包含錯誤型別和錯誤訊息。ResponseSuccess 類別用於表示成功的回應，包含傳回值。build_response_from_invalid_request 函式可以根據無效的請求物件構建 ResponseFailure 物件。
透過上述改進，我們可以更有效地管理錯誤，提升應用程式的健壯性和可維護性。清晰的錯誤訊息和一致的回應結構，也有助於提升使用者經驗。
graph LR
B[B]
Failure[Failure]
Invalid[Invalid]
Request[Request]
Success[Success]
Valid[Valid]
A[Client Request] --> B{build_room_list_request}
B -- Valid Request --> C[RoomListValidRequest]
B -- Invalid Request --> D[RoomListInvalidRequest]
C --> E[Use Case]
D --> F[build_response_from_invalid_request]
E -- Success --> G[ResponseSuccess]
E -- Failure --> H[ResponseFailure]
F --> H
G --> I[Client Response]
H --> I
圖表說明: 此流程圖展示了客戶端請求從建立到傳回回應的完整流程，包含了請求驗證和錯誤處理的邏輯。
處理錯誤的藝術：Python 程式碼的錯誤管理策略
在軟體開發過程中，錯誤處理如同太極的陰陽兩面，與程式碼的功能實作同樣重要。一個健壯的應用程式不僅要能正確執行預期功能，更要能優雅地處理各種異常情況，提供使用者友善的錯誤提示，並保護系統免受意外當機。本文將探討如何在 Python 程式碼中實施有效的錯誤管理策略，從錯誤型別定義到具體實作技巧，帶您領略錯誤處理的精妙之處。
定義清晰的錯誤型別
清晰的錯誤型別定義是有效錯誤管理的根本。以 ResponseTypes 類別為例，它定義了幾種常見的錯誤型別，與 HTTP 錯誤碼遙相呼應，這在 Web 框架中尤其重要。PARAMETERS_ERROR 表示請求引數錯誤，例如使用者輸入了無效的資料。RESOURCE_ERROR 表示請求的資源不可用，例如嘗試讀取不存在的檔案。SYSTEM_ERROR 則表示系統內部錯誤，通常由 Python 程式碼中的異常引起。
graph LR
B[B]
E[E]
Invalid[Invalid]
No[No]
Valid[Valid]
Yes[Yes]
A[Request] --> B{Parameter Validation};
B -- Valid --> C[Process Request];
B -- Invalid --> D[PARAMETERS_ERROR];
C --> E{Resource Available?};
E -- Yes --> F[ResponseSuccess];
E -- No --> G[RESOURCE_ERROR];
C --> H[SYSTEM_ERROR];
內容解密： 上圖展示了請求處理流程以及不同錯誤型別的觸發時機。首先，程式碼會驗證請求引數，若無效則傳回 PARAMETERS_ERROR。若引數有效，則繼續處理請求，並檢查資源是否可用。若資源不可用，則傳回 RESOURCE_ERROR。若在處理過程中發生任何系統錯誤，則傳回 SYSTEM_ERROR。
實作案例：Room List Use Case
讓我們透過 room_list_use_case 函式來看看如何在實際案例中應用錯誤管理。
from rentomatic.responses import (
ResponseSuccess,
ResponseFailure,
ResponseTypes,
build_response_from_invalid_request,
)
def room_list_use_case(repo, request):
if not request:
return build_response_from_invalid_request(request)
try:
rooms = repo.list(filters=request.filters)
return ResponseSuccess(rooms)
except Exception as exc:
return ResponseFailure(ResponseTypes.SYSTEM_ERROR, exc)
內容解密： 這個函式首先檢查 request 是否有效。若無效，則使用 build_response_from_invalid_request 建立一個 ResponseFailure 物件並傳回。若 request 有效，則嘗試從儲存函式庫中取得房間列表。若成功，則傳回一個 ResponseSuccess 物件。若在取得過程中發生異常，則捕捉異常並傳回一個 ResponseFailure 物件，錯誤型別為 SYSTEM_ERROR。
整合外部系統與整合測試
Mocks 在單元測試中非常有用，但它們無法完全模擬真實的外部系統。因此，整合測試至關重要。整合測試可以驗證系統各個元件之間的互動是否正常，並及早發現潛在問題。
以 Flask 開發伺服器為例，它可以作為一個簡單的整合測試環境。當我們修改了 use case 的 API 後，如果沒有相應地更新 REST endpoint，Flask 伺服器就會丟擲錯誤。這凸顯了整合測試的重要性，它能幫助我們發現單元測試中難以察覺的問題。
@mock.patch("application.rest.room.room_list_use_case")
def test_get(mock_use_case, client):
mock_use_case.return_value = ResponseSuccess(rooms)
http_response = client.get("/rooms")
assert json.loads(http_response.data.decode("UTF-8")) == [room_dict]
mock_use_case.assert_called()
args, kwargs = mock_use_case.call_args
assert args[1].filters == {}
assert http_response.status_code == 200
assert http_response.mimetype == "application/json"
內容解密： 這段程式碼展示了一個使用 mock 進行測試的例子。它模擬了 room_list_use_case 的行為，並驗證了 HTTP 伺服器的回應。
這個案例說明瞭錯誤管理的重要性，以及如何在 Python 程式碼中實施有效的錯誤處理策略。透過定義清晰的錯誤型別、使用 try-except 塊處理異常，並進行充分的測試，我們可以構建更健壯、更可靠的應用程式。
總結：有效的錯誤管理是構建健壯應用程式的關鍵。本文探討了錯誤型別定義、實作技巧以及整合測試的重要性，並透過 room_list_use_case 案例展示瞭如何在實踐中應用這些概念。希望這些資訊能幫助您更好地理解和應用錯誤管理策略，提升程式碼品質。
最佳化 Room Listing API：開發更強大的過濾系統
在先前的文章中，我們探討瞭如何建構一個基礎的 Room Listing API。現在，我們將進一步強化這個 API，使其能夠處理更複雜的查詢，並提升其效能和可維護性。
重新設計過濾策略：從應用邏輯到儲存層
最初的設計將過濾邏輯放在應用層，由 use case 處理。然而，資料函式庫通常更擅長處理過濾和排序操作。因此，我決定將過濾邏輯下移至儲存層，以提升效能。
這個調整也讓我們的架構更具彈性。未來如果需要更換儲存系統（例如從記憶體儲存換成資料函式庫），只需修改儲存層的實作，應用層的程式碼可以保持不變。
from rentomatic.domain.room import Room
class MemRepo:
def __init__(self, data):
self.data = data
def list(self, filters=None):
result = [Room.from_dict(i) for i in self.data]
if filters is None:
return result
if "code__eq" in filters:
result = [r for r in result if r.code == filters["code__eq"]]
if "price__eq" in filters:
result = [r for r in result if r.price == int(filters["price__eq"])]
if "price__lt" in filters:
result = [r for r in result if r.price < int(filters["price__lt"])]
if "price__gt" in filters:
result = [r for r in result if r.price > int(filters["price__gt"])]
return result
內容解密：
這段程式碼定義了 MemRepo 類別，負責在記憶體中儲存和查詢房間資料。list 方法接受一個 filters 引數，它是一個字典，用於指定過濾條件。程式碼會根據 filters 中的條件，逐一過濾房間列表，最終傳回符合條件的房間。
我特別注意將價格相關的過濾條件值轉換為整數，避免型別比較錯誤。此外，這個實作也為未來擴充套件更多過濾條件提供了基礎。
更新 API 端點：支援查詢字串
修改儲存層後，我們需要更新 API 端點，使其能夠接收查詢字串，並將其轉換為 filters 引數傳遞給 room_list_use_case。
import json
from flask import Blueprint, request, Response
from rentomatic.repository.memrepo import MemRepo
from rentomatic.use_cases.room_list import room_list_use_case
from rentomatic.serializers.room import RoomJsonEncoder
from rentomatic.requests.room_list import build_room_list_request
from rentomatic.responses import ResponseTypes
blueprint = Blueprint("room", __name__)
STATUS_CODES = {
ResponseTypes.SUCCESS: 200,
ResponseTypes.RESOURCE_ERROR: 404,
ResponseTypes.PARAMETERS_ERROR: 400,
ResponseTypes.SYSTEM_ERROR: 500,
}
rooms = [...] # 省略房間資料
@blueprint.route("/rooms", methods=["GET"])
def room_list():
qrystr_params = {"filters": {}}
for arg, values in request.args.items():
if arg.startswith("filter_"):
qrystr_params["filters"][arg.replace("filter_", "")] = values
request_object = build_room_list_request(filters=qrystr_params["filters"])
repo = MemRepo(rooms)
response = room_list_use_case(repo, request_object)
return Response(
json.dumps(response.value, cls=RoomJsonEncoder),
mimetype="application/json",
status=STATUS_CODES[response.type],
)
內容解密：
這段程式碼定義了 /rooms 端點的處理邏輯。它會解析查詢字串，將 filter_ 開頭的引數轉換為過濾條件，然後建構 request_object 並傳遞給 room_list_use_case。最後，它會將 use case 的執行結果轉換為 JSON 格式，並傳回給使用者端。
測試策略調整：驗證過濾功能
為了確保新的過濾功能正常運作，我新增了多個測試案例，涵蓋不同的過濾條件和組合。
# ... 省略其他測試案例
def test_repository_list_with_price_between_filter(room_dicts):
repo = MemRepo(room_dicts)
rooms = repo.list(filters={"price__lt": 66, "price__gt": 48})
assert len(rooms) == 1
assert rooms[0].code == "913694c6-435a-4366-ba0d-da5334a611b2"
內容解密：
這個測試案例驗證了使用 price__lt 和 price__gt 兩個過濾條件的組合查詢。它確保只有價格在 48 到 66 之間的房間會被傳回。
透過這些改進，我們的 Room Listing API 不僅支援更強大的過濾功能，而與效能和可維護性也得到了提升。這也為我們未來整合更複雜的查詢和儲存系統打下了堅實的基礎。
graph LR
B[B]
A[Client] --> B{API Endpoint}
B --> C[Request Object]
C --> D[Use Case]
D --> E[Repository]
E --> F[Data Storage]
F --> D
D --> C
C --> B
B --> A[Response]
圖表說明: 此流程圖展示了使用者端請求 /rooms API，經過 API 端點處理、建構請求物件、執行 Use Case、查詢儲存函式庫，最後傳回回應的完整流程。
classDiagram
class Room {
-code: string
-size: int
-price: int
-longitude: float
-latitude: float
+from_dict(data: dict): Room
}
class MemRepo {
-data: list
+list(filters: dict): list[Room]
}
圖表說明: 此類別圖展示了 Room 和 MemRepo 類別的結構和屬性。MemRepo 的 list 方法接受過濾條件，並傳回符合條件的 Room 物件列表。
在先前的文章中，我們使用記憶體資料函式庫展示了儲存函式庫層抽象化的概念。然而，記憶體資料函式庫不適用於正式環境，因此需要整合更穩固的儲存方案，例如PostgreSQL。本文將引導你使用PostgreSQL建立儲存函式庫，並透過整合測試驗證應用程式與資料函式庫的正確互動。
Clean Architecture 的解耦合優勢
Clean Architecture 的核心優勢之一在於元件替換的簡便性。我們先前設計的 use case 接收儲存函式庫例項作為引數，並使用其 list 方法檢索資料。這種設計讓 use case 與儲存函式庫之間的耦合度非常低，僅透過物件提供的 API 連線，而非具體實作。換句話說，use case 對於 list 方法是多型的。
這種鬆散耦合意味著，只要新實作提供所需的介面，我們可以隨時用不同的實作替換 use case 和儲存函式庫。例如，儲存函式庫的初始化並非 use case 使用的 API 的一部分，因為儲存函式庫是在主程式碼中初始化，而不是在每個 use case 中。因此，__init__ 方法在不同的儲存函式庫實作中不需要相同，這提供了很大的靈活性，因為不同的儲存系統可能需要不同的初始化值。
先前實作的簡單儲存函式庫如下：
from rentomatic.domain.room import Room
class MemRepo:
def __init__(self, data):
self.data = data
def list(self, filters=None):
result = [Room.from_dict(i) for i in self.data]
if filters is None:
return result
if "code__eq" in filters:
result = [r for r in result if r.code == filters["code__eq"]]
if "price__eq" in filters:
result = [r for r in result if r.price == int(filters["price__eq"])]
if "price__lt" in filters:
result = [r for r in result if r.price < int(filters["price__lt"])]
if "price__gt" in filters:
result = [r for r in result if r.price > int(filters["price__gt"])]
return result
這個儲存函式庫的介面由兩部分組成：初始化和 list 方法。__init__ 方法接受值，因為這個物件並非長期儲存，每次例項化類別時都必須傳遞資料。根據資料函式庫的儲存函式庫不需要在初始化時填入資料，其主要工作是在不同 session 之間儲存資料，但仍然需要至少使用資料函式庫地址和存取憑證進行初始化。
根據 PostgreSQL 的儲存函式庫
現在，我們將使用 PostgreSQL 建立儲存函式庫，並透過 SQLAlchemy 介面與 Python 互動。SQLAlchemy 是一個 ORM，它將物件對映到關聯式資料函式庫。ORM 通常可以在 Web 框架（如 Django）或獨立套件中找到。
重要的是，ORM 不適合使用 mock 測試。對 SQLAlchemy 查詢結構進行 mock 會導致程式碼非常複雜，難以編寫和維護，因為查詢的每個更改都會導致一系列 mock 需要重新編寫。
因此，我們需要設定整合測試。其概念是建立資料函式庫，設定與 SQLAlchemy 的連線，測試所需條件，然後銷毀資料函式庫。由於建立和銷毀資料函式庫的操作可能很耗時，我們可能只想在整個測試套件的開始和結束時執行一次。即使如此，測試仍然會比較慢，因此我們需要使用標籤來避免每次執行測試套件時都執行這些測試。
標記整合測試
首先，我們需要標記整合測試，預設情況下排除它們，並建立執行它們的方法。由於 pytest 支援標記（稱為 marks），我們可以使用此功能將全域標記增加到整個模組。建立檔案 tests/repository/postgres/test_postgresrepo.py 並放入以下程式碼：
import pytest
pytestmark = pytest.mark.integration
內容解密：
這段程式碼匯入了 pytest 函式庫，並使用 pytestmark 變數將 pytest.mark.integration 標記應用於整個模組。這意味著模組中的所有測試函式都將自動標記為整合測試。
接下來，我們將在下一篇文章中繼續探討如何設定 PostgreSQL 資料函式庫以及如何編寫整合測試。
