完全な Web API クライアントの構築
この最終ステップでは、学習したすべてのことを組み合わせて、堅牢なエラーハンドリングを備えた完全な Web API クライアントを構築します。テストに使用できる無料のオンライン REST API である JSONPlaceholder API のクライアントを作成します。
簡単な API クライアントの作成
http_status_lab ディレクトリに、api_client.py という名前のファイルを作成しましょう。
import requests
import json
import logging
## Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class APIClient:
"""A simple API client with robust error handling."""
def __init__(self, base_url):
"""Initialize the API client with a base URL."""
self.base_url = base_url
self.session = requests.Session()
def request(self, endpoint, method="get", params=None, data=None, headers=None):
"""Make a request to the API with error handling."""
url = f"{self.base_url}{endpoint}"
## Default headers for JSON APIs
if headers is None:
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
## Convert data to JSON if it's a dictionary
json_data = None
if data and isinstance(data, dict):
json_data = data
data = None
try:
## Log the request
logger.info(f"Making {method.upper()} request to {url}")
## Make the request
response = self.session.request(
method=method,
url=url,
params=params,
data=data,
json=json_data,
headers=headers
)
## Log the response status
logger.info(f"Received response with status code {response.status_code}")
## Check for HTTP errors
response.raise_for_status()
## Parse JSON response if applicable
try:
return response.json()
except json.JSONDecodeError:
return response.text
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
error_message = f"HTTP Error: {status_code}"
try:
## Try to get more details from the response
error_data = e.response.json()
error_message = f"{error_message} - {error_data.get('message', str(error_data))}"
except (json.JSONDecodeError, AttributeError):
error_message = f"{error_message} - {e.response.text if hasattr(e, 'response') else str(e)}"
logger.error(error_message)
## Re-raise with more context
if 400 <= status_code < 500:
logger.error(f"Client error: {error_message}")
raise Exception(f"Client error (HTTP {status_code}): {error_message}")
elif 500 <= status_code < 600:
logger.error(f"Server error: {error_message}")
raise Exception(f"Server error (HTTP {status_code}): {error_message}")
else:
raise
except requests.exceptions.ConnectionError as e:
logger.error(f"Connection Error: {e}")
raise Exception(f"Connection Error: Could not connect to {url}")
except requests.exceptions.Timeout as e:
logger.error(f"Timeout Error: {e}")
raise Exception(f"Timeout Error: Request to {url} timed out")
except requests.exceptions.RequestException as e:
logger.error(f"Request Exception: {e}")
raise Exception(f"Request Error: {str(e)}")
def get(self, endpoint, params=None, headers=None):
"""Make a GET request to the API."""
return self.request(endpoint, method="get", params=params, headers=headers)
def post(self, endpoint, data=None, headers=None):
"""Make a POST request to the API."""
return self.request(endpoint, method="post", data=data, headers=headers)
def put(self, endpoint, data=None, headers=None):
"""Make a PUT request to the API."""
return self.request(endpoint, method="put", data=data, headers=headers)
def delete(self, endpoint, headers=None):
"""Make a DELETE request to the API."""
return self.request(endpoint, method="delete", headers=headers)
## Test the API client with JSONPlaceholder
def test_api_client():
"""Test the API client with JSONPlaceholder."""
client = APIClient("https://jsonplaceholder.typicode.com")
try:
## Get a list of posts
print("\nGetting posts:")
posts = client.get("/posts")
print(f"Retrieved {len(posts)} posts")
print(f"First post: {posts[0]}")
## Get a single post
print("\nGetting a single post:")
post = client.get("/posts/1")
print(f"Retrieved post: {post}")
## Get a non-existent post (404)
print("\nTrying to get a non-existent post:")
try:
client.get("/posts/999999")
except Exception as e:
print(f"Expected error: {e}")
## Create a new post
print("\nCreating a new post:")
new_post = client.post("/posts", {
"title": "My New Post",
"body": "This is the content of my new post.",
"userId": 1
})
print(f"Created post: {new_post}")
## Update a post
print("\nUpdating a post:")
updated_post = client.put("/posts/1", {
"id": 1,
"title": "Updated Title",
"body": "Updated content.",
"userId": 1
})
print(f"Updated post: {updated_post}")
## Delete a post
print("\nDeleting a post:")
delete_response = client.delete("/posts/1")
print(f"Delete response: {delete_response}")
except Exception as e:
print(f"An error occurred: {e}")
if __name__ == "__main__":
test_api_client()
スクリプトを実行して、完全な API クライアントが様々なシナリオをどのように処理するかを確認します。
python api_client.py
エラーハンドリングのテスト
API クライアントのエラーハンドリング機能をテストするための別のファイルを作成しましょう。http_status_lab ディレクトリに、test_error_handling.py という名前のファイルを作成します。
from api_client import APIClient
import time
def test_different_errors():
"""Test different error scenarios with our API client."""
## Test with a valid API
print("\n1. Testing with a valid API:")
valid_client = APIClient("https://jsonplaceholder.typicode.com")
try:
data = valid_client.get("/posts/1")
print(f"Success! Retrieved data: {data}")
except Exception as e:
print(f"Unexpected error: {e}")
## Test with a 404 error
print("\n2. Testing with a non-existent endpoint (404):")
try:
valid_client.get("/non_existent_endpoint")
except Exception as e:
print(f"Expected error: {e}")
## Test with an invalid host
print("\n3. Testing with an invalid host (Connection Error):")
invalid_client = APIClient("https://this-does-not-exist-123456789.com")
try:
invalid_client.get("/anything")
except Exception as e:
print(f"Expected error: {e}")
## Test with a timeout
print("\n4. Testing with a timeout:")
timeout_client = APIClient("https://httpbin.org")
try:
## httpbin.org/delay/5 will delay the response for 5 seconds
## but we'll set the timeout to 2 seconds
timeout_client.session.request = lambda **kwargs: timeout_client.session.request_original(
**{**kwargs, 'timeout': 2}
)
timeout_client.session.request_original = timeout_client.session.request
timeout_client.get("/delay/5")
except Exception as e:
print(f"Expected error: {e}")
if __name__ == "__main__":
test_different_errors()
上記のスクリプトは、実行時にメソッドを変更しようとしているため、タイムアウト処理が原因でエラーが発生する可能性があります。これらの問題を回避するために、これを簡素化しましょう。
from api_client import APIClient
def test_different_errors():
"""Test different error scenarios with our API client."""
## Test with a valid API
print("\n1. Testing with a valid API:")
valid_client = APIClient("https://jsonplaceholder.typicode.com")
try:
data = valid_client.get("/posts/1")
print(f"Success! Retrieved data: {data}")
except Exception as e:
print(f"Unexpected error: {e}")
## Test with a 404 error
print("\n2. Testing with a non-existent endpoint (404):")
try:
valid_client.get("/non_existent_endpoint")
except Exception as e:
print(f"Expected error: {e}")
## Test with a server error
print("\n3. Testing with a server error (500):")
error_client = APIClient("https://httpbin.org")
try:
error_client.get("/status/500")
except Exception as e:
print(f"Expected error: {e}")
if __name__ == "__main__":
test_different_errors()
エラーテストスクリプトを実行します。
python test_error_handling.py
このスクリプトは、API クライアントが様々なエラーシナリオをどのように処理するかを示し、実際のアプリケーションのための堅牢な基盤を提供します。
主なポイント
この完全な API クライアントを構築することで、いくつかの重要な技術を学びました。
- 再利用可能な API クライアントクラスの作成
- 包括的なエラーハンドリングの実装
- デバッグのためのリクエストとレスポンスのロギング
- 様々なレスポンス形式の解析
- 様々な HTTP メソッド (GET、POST、PUT、DELETE) の処理
これらのスキルは、Web API と対話する堅牢な Python アプリケーションを構築するために不可欠であり、コードが様々なエラーシナリオを適切に処理し、ユーザーに意味のあるエラーメッセージを提供できるようにします。