Python 字典與外部 API 交互的 23 個模式
作者：小白PythonAI編程
API（Application Programming Interface）是應(yīng)用程序之間通信的接口，而Python字典是一種非常靈活的數(shù)據(jù)結(jié)構(gòu)，非常適合處理API返回的數(shù)據(jù)。
大家好！今天我們要聊的是如何使用Python字典與外部API進行交互。API（Application Programming Interface）是應(yīng)用程序之間通信的接口，而Python字典是一種非常靈活的數(shù)據(jù)結(jié)構(gòu)，非常適合處理API返回的數(shù)據(jù)。我們將從簡單的概念開始，逐步深入到更高級的技術(shù)，幫助你更好地理解和掌握這些技能。
1. 發(fā)送GET請求并解析JSON響應(yīng)
首先，我們來看看如何發(fā)送GET請求并解析JSON響應(yīng)。我們將使用requests庫來發(fā)送請求，并將響應(yīng)解析為字典。
import requests
# 發(fā)送GET請求
response = requests.get('https://api.example.com/data')
# 檢查請求是否成功
if response.status_code == 200:
# 將響應(yīng)解析為字典
data = response.json()
print(data)
else:
print(f"請求失敗，狀態(tài)碼: {response.status_code}")解釋：
- requests.get發(fā)送GET請求。
- response.status_code檢查HTTP狀態(tài)碼，200表示成功。
- response.json()將響應(yīng)體解析為字典。
2. 處理嵌套字典
API返回的數(shù)據(jù)往往包含嵌套字典。我們需要學會如何訪問嵌套數(shù)據(jù)。
data = {
"user": {
"name": "Alice",
"age": 30,
"address": {
"street": "123 Main St",
"city": "New York"
}
}
}
# 訪問嵌套數(shù)據(jù)
user_name = data['user']['name']
user_city = data['user']['address']['city']
print(f"用戶姓名: {user_name}, 城市: {user_city}")解釋：
- 使用多重索引訪問嵌套字典中的值。
3. 遍歷字典
有時候我們需要遍歷字典中的所有鍵值對?？梢允褂胒or循環(huán)來實現(xiàn)。
data = {
"name": "Alice",
"age": 30,
"city": "New York"
}
# 遍歷字典
for key, value in data.items():
print(f"{key}: {value}")解釋：
- data.items()返回一個包含鍵值對的列表。
- for循環(huán)遍歷每個鍵值對。
4. 處理列表中的字典
API返回的數(shù)據(jù)中可能包含列表，列表中的每個元素都是一個字典。我們需要學會如何處理這種情況。
data = [
{"name": "Alice", "age": 30},
{"name": "Bob", "age": 25}
]
# 遍歷列表中的字典
for user in data:
print(f"姓名: {user['name']}, 年齡: {user['age']}")解釋：
- for循環(huán)遍歷列表中的每個字典。
5. 添加和修改字典項
我們可以使用字典的update方法或直接賦值來添加和修改字典項。
data = {
"name": "Alice",
"age": 30
}
# 添加新項
data['email'] = 'alice@example.com'
# 修改現(xiàn)有項
data.update({"age": 31})
print(data)解釋：
- data['email'] = 'alice@example.com'添加新項。
- data.update({"age": 31})修改現(xiàn)有項。
6. 刪除字典項
使用del關(guān)鍵字或pop方法可以刪除字典中的項。
data = {
"name": "Alice",
"age": 30,
"email": "alice@example.com"
}
# 刪除項
del data['email']
age = data.pop('age')
print(data)
print(f"刪除的年齡: {age}")解釋：
- del data['email']刪除指定鍵的項。
- data.pop('age')刪除并返回指定鍵的值。
7. 檢查字典中是否存在鍵
使用in關(guān)鍵字可以檢查字典中是否存在某個鍵。
data = {
"name": "Alice",
"age": 30
}
# 檢查鍵是否存在
if 'email' in data:
print("存在email")
else:
print("不存在email")
if 'name' in data:
print("存在name")解釋：
- if 'email' in data檢查字典中是否存在email鍵。
8. 獲取字典的長度
使用len函數(shù)可以獲取字典的長度。
data = {
"name": "Alice",
"age": 30
}
# 獲取字典長度
length = len(data)
print(f"字典長度: {length}")解釋：
- len(data)返回字典中鍵值對的數(shù)量。
9. 獲取字典的所有鍵和值
使用keys和values方法可以分別獲取字典的所有鍵和值。
data = {
"name": "Alice",
"age": 30
}
# 獲取所有鍵
keys = data.keys()
print(f"所有鍵: {list(keys)}")
# 獲取所有值
values = data.values()
print(f"所有值: {list(values)}")解釋：
- data.keys()返回所有鍵的視圖。
- data.values()返回所有值的視圖。
10. 使用字典推導式
字典推導式是一種簡潔的方式來創(chuàng)建字典。
data = ["Alice", "Bob", "Charlie"]
# 字典推導式
user_dict = {name: len(name) for name in data}
print(user_dict)解釋：
- {name: len(name) for name in data}創(chuàng)建一個新的字典，鍵為名字，值為名字的長度。
11. 處理API錯誤和異常
在與API交互時，可能會遇到各種錯誤和異常。我們需要學會如何處理這些情況。
import requests
try:
response = requests.get('https://api.example.com/data')
response.raise_for_status() # 如果響應(yīng)狀態(tài)碼不是200，拋出HTTPError
data = response.json()
print(data)
except requests.exceptions.HTTPError as errh:
print(f"HTTP Error: {errh}")
except requests.exceptions.ConnectionError as errc:
print(f"Error Connecting: {errc}")
except requests.exceptions.Timeout as errt:
print(f"Timeout Error: {errt}")
except requests.exceptions.RequestException as err:
print(f"Something went wrong: {err}")解釋：
- response.raise_for_status()檢查響應(yīng)狀態(tài)碼，如果不是200，拋出HTTPError。
- 使用try-except塊捕獲并處理各種異常。
12. 使用環(huán)境變量存儲API密鑰
為了安全起見，我們通常不希望將API密鑰硬編碼在代碼中?？梢允褂铆h(huán)境變量來存儲API密鑰。
import os
import requests
# 獲取環(huán)境變量
api_key = os.getenv('API_KEY')
# 發(fā)送請求
response = requests.get(f'https://api.example.com/data?api_key={api_key}')
data = response.json()
print(data)解釋：
- os.getenv('API_KEY')獲取環(huán)境變量API_KEY的值。
13. 處理分頁數(shù)據(jù)
許多API返回的數(shù)據(jù)是分頁的。我們需要學會如何處理分頁數(shù)據(jù)。
import requests
def fetch_data(page):
response = requests.get(f'https://api.example.com/data?page={page}')
return response.json()
# 獲取第一頁數(shù)據(jù)
data = fetch_data(1)
# 檢查是否有更多頁面
while 'next_page' in data and data['next_page']:
next_page = data['next_page']
data = fetch_data(next_page)
print(data)解釋：
- fetch_data(page)函數(shù)發(fā)送請求并返回指定頁面的數(shù)據(jù)。
- 使用while循環(huán)檢查是否有更多頁面，并繼續(xù)獲取數(shù)據(jù)。
14. 使用會話管理器
requests庫提供了會話管理器，可以提高與API交互的效率。
import requests
# 創(chuàng)建會話
session = requests.Session()
# 發(fā)送多個請求
response1 = session.get('https://api.example.com/data1')
response2 = session.get('https://api.example.com/data2')
data1 = response1.json()
data2 = response2.json()
print(data1)
print(data2)解釋：
- requests.Session()創(chuàng)建一個會話對象。
- 使用會話對象發(fā)送多個請求，會話對象會自動管理連接池。
15. 處理API認證
許多API需要認證才能訪問。我們可以使用requests庫提供的認證機制。
import requests
from requests.auth import HTTPBasicAuth
# 發(fā)送帶有基本認證的請求
response = requests.get('https://api.example.com/data', auth=HTTPBasicAuth('username', 'password'))
data = response.json()
print(data)解釋：
- HTTPBasicAuth('username', 'password')創(chuàng)建基本認證對象。
- auth參數(shù)傳遞認證對象。
16. 使用異步請求
對于需要高并發(fā)的場景，可以使用aiohttp庫發(fā)送異步請求。
import aiohttp
import asyncio
async def fetch_data(session, url):
async with session.get(url) as response:
return await response.json()
async def main():
async with aiohttp.ClientSession() as session:
tasks = [
fetch_data(session, 'https://api.example.com/data1'),
fetch_data(session, 'https://api.example.com/data2')
]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
# 運行異步主函數(shù)
asyncio.run(main())在上一部分中，我們介紹了如何使用Python字典與外部API進行基本的交互，包括發(fā)送GET請求、處理嵌套字典、遍歷字典、添加和修改字典項等?，F(xiàn)在，我們將繼續(xù)深入探討更高級的概念和技術(shù)。
17. 使用POST請求發(fā)送數(shù)據(jù)
有時候我們需要向API發(fā)送數(shù)據(jù)，這通常通過POST請求來實現(xiàn)。我們可以使用requests庫的post方法來發(fā)送POST請求。
import requests
# 定義要發(fā)送的數(shù)據(jù)
data = {
"name": "Alice",
"age": 30
}
# 發(fā)送POST請求
response = requests.post('https://api.example.com/create_user', json=data)
# 檢查請求是否成功
if response.status_code == 201:
print("用戶創(chuàng)建成功")
created_user = response.json()
print(created_user)
else:
print(f"請求失敗，狀態(tài)碼: {response.status_code}")解釋：
- requests.post發(fā)送POST請求。
- json=data將字典轉(zhuǎn)換為JSON格式并發(fā)送。
- response.json()解析響應(yīng)體為字典。
18. 處理復(fù)雜的API響應(yīng)
有些API返回的響應(yīng)可能非常復(fù)雜，包含多個嵌套層級。我們需要學會如何處理這些復(fù)雜的響應(yīng)。
import requests
# 發(fā)送GET請求
response = requests.get('https://api.example.com/complex_data')
# 檢查請求是否成功
if response.status_code == 200:
data = response.json()
# 處理嵌套數(shù)據(jù)
users = data['users']
for user in users:
name = user['name']
address = user['address']
city = address['city']
print(f"姓名: {name}, 城市: {city}")
else:
print(f"請求失敗，狀態(tài)碼: {response.status_code}")解釋：
- data['users']訪問嵌套的用戶列表。
- user['address']['city']訪問嵌套的地址信息。
19. 使用類封裝API交互
為了提高代碼的可維護性和復(fù)用性，可以使用類來封裝API交互邏輯。
import requests
class APIClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.api_key = api_key
self.session = requests.Session()
def get(self, endpoint):
url = f"{self.base_url}/{endpoint}?api_key={self.api_key}"
response = self.session.get(url)
response.raise_for_status()
return response.json()
def post(self, endpoint, data):
url = f"{self.base_url}/{endpoint}?api_key={self.api_key}"
response = self.session.post(url, json=data)
response.raise_for_status()
return response.json()
# 使用API客戶端
client = APIClient('https://api.example.com', os.getenv('API_KEY'))
# 獲取數(shù)據(jù)
data = client.get('data')
print(data)
# 發(fā)送數(shù)據(jù)
response = client.post('create_user', {'name': 'Alice', 'age': 30})
print(response)解釋：
- APIClient類封裝了API交互邏輯。
- __init__方法初始化API客戶端。
- get和post方法分別發(fā)送GET和POST請求。
20. 使用緩存優(yōu)化性能
頻繁請求API可能會導致性能問題，可以使用緩存來優(yōu)化性能。我們可以使用requests_cache庫來實現(xiàn)緩存。
import requests_cache
import requests
# 啟用緩存
requests_cache.install_cache('api_cache', backend='sqlite', expire_after=3600)
# 發(fā)送請求
response = requests.get('https://api.example.com/data')
# 檢查請求是否成功
if response.status_code == 200:
data = response.json()
print(data)
else:
print(f"請求失敗，狀態(tài)碼: {response.status_code}")解釋：
- requests_cache.install_cache啟用緩存，指定緩存后端和過期時間。
- 緩存會在第一次請求時存儲數(shù)據(jù)，后續(xù)請求會直接從緩存中讀取。
21. 使用OAuth2進行認證
許多現(xiàn)代API使用OAuth2進行認證。我們可以使用requests-oauthlib庫來處理OAuth2認證。
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import BackendApplicationClient
import os
# 定義客戶端ID和密鑰
client_id = os.getenv('CLIENT_ID')
client_secret = os.getenv('CLIENT_SECRET')
# 創(chuàng)建OAuth2客戶端
client = BackendApplicationClient(client_id=client_id)
oauth = OAuth2Session(client=client)
# 獲取訪問令牌
token = oauth.fetch_token(token_url='https://api.example.com/oauth/token', client_id=client_id, client_secret=client_secret)
# 發(fā)送請求
response = oauth.get('https://api.example.com/data')
# 檢查請求是否成功
if response.status_code == 200:
data = response.json()
print(data)
else:
print(f"請求失敗，狀態(tài)碼: {response.status_code}")解釋：
- BackendApplicationClient創(chuàng)建OAuth2客戶端。
- oauth.fetch_token獲取訪問令牌。
- 使用oauth.get發(fā)送帶有訪問令牌的請求。
22. 處理API限流
許多API有請求頻率限制，我們需要學會如何處理這些限制。可以使用time.sleep來控制請求頻率。
import requests
import time
# 定義請求間隔時間
request_interval = 1 # 每秒最多發(fā)送一次請求
def fetch_data(page):
response = requests.get(f'https://api.example.com/data?page={page}')
return response.json()
# 獲取第一頁數(shù)據(jù)
data = fetch_data(1)
# 檢查是否有更多頁面
while 'next_page' in data and data['next_page']:
next_page = data['next_page']
time.sleep(request_interval) # 控制請求頻率
data = fetch_data(next_page)
print(data)解釋：
- time.sleep(request_interval)控制每次請求之間的間隔時間。
23. 使用Web框架集成API
在Web應(yīng)用中，我們經(jīng)常需要集成API。可以使用Flask框架來實現(xiàn)。
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
@app.route('/get_weather', methods=['GET'])
def get_weather():
city = request.args.get('city')
api_key = os.getenv('WEATHER_API_KEY')
url = f'https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric'
try:
response = requests.get(url)
response.raise_for_status()
data = response.json()
if data['cod'] == 200:
weather = data['weather'][0]['description']
temperature = data['main']['temp']
return jsonify({
"city": city,
"weather": weather,
"temperature": temperature
})
else:
return jsonify({"error": data['message']}), 400
except requests.exceptions.HTTPError as errh:
return jsonify({"error": str(errh)}), 500
except requests.exceptions.ConnectionError as errc:
return jsonify({"error": str(errc)}), 500
except requests.exceptions.Timeout as errt:
return jsonify({"error": str(errt)}), 500
except requests.exceptions.RequestException as err:
return jsonify({"error": str(err)}), 500
if __name__ == '__main__':
app.run(debug=True)解釋：
- Flask創(chuàng)建一個Web應(yīng)用。
- @app.route定義路由。
- request.args.get('city')獲取查詢參數(shù)。
- jsonify返回JSON響應(yīng)。
責任編輯：趙寧寧
來源：
小白PythonAI編程
