Python 字典與外部 API 交互的 23 個(gè)模式

作者：小白PythonAI編程 2024-10-29 16:13:04

API（Application Programming Interface）是應(yīng)用程序之間通信的接口，而Python字典是一種非常靈活的數(shù)據(jù)結(jié)構(gòu)，非常適合處理API返回的數(shù)據(jù)。

大家好！今天我們要聊的是如何使用Python字典與外部API進(jìn)行交互。API（Application Programming Interface）是應(yīng)用程序之間通信的接口，而Python字典是一種非常靈活的數(shù)據(jù)結(jié)構(gòu)，非常適合處理API返回的數(shù)據(jù)。我們將從簡(jiǎn)單的概念開(kāi)始，逐步深入到更高級(jí)的技術(shù)，幫助你更好地理解和掌握這些技能。

1. 發(fā)送GET請(qǐng)求并解析JSON響應(yīng)

首先，我們來(lái)看看如何發(fā)送GET請(qǐng)求并解析JSON響應(yīng)。我們將使用requests庫(kù)來(lái)發(fā)送請(qǐng)求，并將響應(yīng)解析為字典。

import requests

# 發(fā)送GET請(qǐng)求
response = requests.get('https://api.example.com/data')

# 檢查請(qǐng)求是否成功
if response.status_code == 200:
    # 將響應(yīng)解析為字典
    data = response.json()
    print(data)
else:
    print(f"請(qǐng)求失敗，狀態(tài)碼: {response.status_code}")

解釋：

requests.get發(fā)送GET請(qǐng)求。
response.status_code檢查HTTP狀態(tài)碼，200表示成功。
response.json()將響應(yīng)體解析為字典。

2. 處理嵌套字典

API返回的數(shù)據(jù)往往包含嵌套字典。我們需要學(xué)會(huì)如何訪問(wèn)嵌套數(shù)據(jù)。

data = {
    "user": {
        "name": "Alice",
        "age": 30,
        "address": {
            "street": "123 Main St",
            "city": "New York"
        }
    }
}

# 訪問(wèn)嵌套數(shù)據(jù)
user_name = data['user']['name']
user_city = data['user']['address']['city']

print(f"用戶姓名: {user_name}, 城市: {user_city}")

解釋：

使用多重索引訪問(wèn)嵌套字典中的值。

3. 遍歷字典

有時(shí)候我們需要遍歷字典中的所有鍵值對(duì)。可以使用for循環(huán)來(lái)實(shí)現(xiàn)。

data = {
    "name": "Alice",
    "age": 30,
    "city": "New York"
}

# 遍歷字典
for key, value in data.items():
    print(f"{key}: {value}")

解釋：

data.items()返回一個(gè)包含鍵值對(duì)的列表。
for循環(huán)遍歷每個(gè)鍵值對(duì)。

4. 處理列表中的字典

API返回的數(shù)據(jù)中可能包含列表，列表中的每個(gè)元素都是一個(gè)字典。我們需要學(xué)會(huì)如何處理這種情況。

data = [
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25}
]

# 遍歷列表中的字典
for user in data:
    print(f"姓名: {user['name']}, 年齡: {user['age']}")

解釋：

for循環(huán)遍歷列表中的每個(gè)字典。

5. 添加和修改字典項(xiàng)

我們可以使用字典的update方法或直接賦值來(lái)添加和修改字典項(xiàng)。

data = {
    "name": "Alice",
    "age": 30
}

# 添加新項(xiàng)
data['email'] = 'alice@example.com'

# 修改現(xiàn)有項(xiàng)
data.update({"age": 31})

print(data)

解釋：

data['email'] = 'alice@example.com'添加新項(xiàng)。
data.update({"age": 31})修改現(xiàn)有項(xiàng)。

6. 刪除字典項(xiàng)

使用del關(guān)鍵字或pop方法可以刪除字典中的項(xiàng)。

data = {
    "name": "Alice",
    "age": 30,
    "email": "alice@example.com"
}

# 刪除項(xiàng)
del data['email']
age = data.pop('age')

print(data)
print(f"刪除的年齡: {age}")

解釋：

del data['email']刪除指定鍵的項(xiàng)。
data.pop('age')刪除并返回指定鍵的值。

7. 檢查字典中是否存在鍵

使用in關(guān)鍵字可以檢查字典中是否存在某個(gè)鍵。

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. 獲取字典的長(zhǎng)度

使用len函數(shù)可以獲取字典的長(zhǎng)度。

data = {
    "name": "Alice",
    "age": 30
}

# 獲取字典長(zhǎng)度
length = len(data)
print(f"字典長(zhǎng)度: {length}")

解釋：

len(data)返回字典中鍵值對(duì)的數(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. 使用字典推導(dǎo)式

字典推導(dǎo)式是一種簡(jiǎn)潔的方式來(lái)創(chuàng)建字典。

data = ["Alice", "Bob", "Charlie"]

# 字典推導(dǎo)式
user_dict = {name: len(name) for name in data}

print(user_dict)

解釋：

{name: len(name) for name in data}創(chuàng)建一個(gè)新的字典，鍵為名字，值為名字的長(zhǎng)度。

11. 處理API錯(cuò)誤和異常

在與API交互時(shí)，可能會(huì)遇到各種錯(cuò)誤和異常。我們需要學(xué)會(huì)如何處理這些情況。

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)境變量存儲(chǔ)API密鑰

為了安全起見(jiàn)，我們通常不希望將API密鑰硬編碼在代碼中。可以使用環(huán)境變量來(lái)存儲(chǔ)API密鑰。

import os
import requests

# 獲取環(huán)境變量
api_key = os.getenv('API_KEY')

# 發(fā)送請(qǐng)求
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. 處理分頁(yè)數(shù)據(jù)

許多API返回的數(shù)據(jù)是分頁(yè)的。我們需要學(xué)會(huì)如何處理分頁(yè)數(shù)據(jù)。

import requests

def fetch_data(page):
    response = requests.get(f'https://api.example.com/data?page={page}')
    return response.json()

# 獲取第一頁(yè)數(shù)據(jù)
data = fetch_data(1)

# 檢查是否有更多頁(yè)面
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ā)送請(qǐng)求并返回指定頁(yè)面的數(shù)據(jù)。
使用while循環(huán)檢查是否有更多頁(yè)面，并繼續(xù)獲取數(shù)據(jù)。

14. 使用會(huì)話管理器

requests庫(kù)提供了會(huì)話管理器，可以提高與API交互的效率。

import requests

# 創(chuàng)建會(huì)話
session = requests.Session()

# 發(fā)送多個(gè)請(qǐng)求
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)建一個(gè)會(huì)話對(duì)象。
使用會(huì)話對(duì)象發(fā)送多個(gè)請(qǐng)求，會(huì)話對(duì)象會(huì)自動(dòng)管理連接池。

15. 處理API認(rèn)證

許多API需要認(rèn)證才能訪問(wèn)。我們可以使用requests庫(kù)提供的認(rèn)證機(jī)制。

import requests
from requests.auth import HTTPBasicAuth

# 發(fā)送帶有基本認(rèn)證的請(qǐng)求
response = requests.get('https://api.example.com/data', auth=HTTPBasicAuth('username', 'password'))

data = response.json()
print(data)

解釋：

HTTPBasicAuth('username', 'password')創(chuàng)建基本認(rèn)證對(duì)象。
auth參數(shù)傳遞認(rèn)證對(duì)象。

16. 使用異步請(qǐng)求

對(duì)于需要高并發(fā)的場(chǎng)景，可以使用aiohttp庫(kù)發(fā)送異步請(qǐng)求。

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)

# 運(yùn)行異步主函數(shù)
asyncio.run(main())

在上一部分中，我們介紹了如何使用Python字典與外部API進(jìn)行基本的交互，包括發(fā)送GET請(qǐng)求、處理嵌套字典、遍歷字典、添加和修改字典項(xiàng)等。現(xiàn)在，我們將繼續(xù)深入探討更高級(jí)的概念和技術(shù)。

17. 使用POST請(qǐng)求發(fā)送數(shù)據(jù)

有時(shí)候我們需要向API發(fā)送數(shù)據(jù)，這通常通過(guò)POST請(qǐng)求來(lái)實(shí)現(xiàn)。我們可以使用requests庫(kù)的post方法來(lái)發(fā)送POST請(qǐng)求。

import requests

# 定義要發(fā)送的數(shù)據(jù)
data = {
    "name": "Alice",
    "age": 30
}

# 發(fā)送POST請(qǐng)求
response = requests.post('https://api.example.com/create_user', json=data)

# 檢查請(qǐng)求是否成功
if response.status_code == 201:
    print("用戶創(chuàng)建成功")
    created_user = response.json()
    print(created_user)
else:
    print(f"請(qǐng)求失敗，狀態(tài)碼: {response.status_code}")

解釋：

requests.post發(fā)送POST請(qǐng)求。
json=data將字典轉(zhuǎn)換為JSON格式并發(fā)送。
response.json()解析響應(yīng)體為字典。

18. 處理復(fù)雜的API響應(yīng)

有些API返回的響應(yīng)可能非常復(fù)雜，包含多個(gè)嵌套層級(jí)。我們需要學(xué)會(huì)如何處理這些復(fù)雜的響應(yīng)。

import requests

# 發(fā)送GET請(qǐng)求
response = requests.get('https://api.example.com/complex_data')

# 檢查請(qǐng)求是否成功
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"請(qǐng)求失敗，狀態(tài)碼: {response.status_code}")

解釋：

data['users']訪問(wèn)嵌套的用戶列表。
user['address']['city']訪問(wèn)嵌套的地址信息。

19. 使用類封裝API交互

為了提高代碼的可維護(hù)性和復(fù)用性，可以使用類來(lái)封裝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請(qǐng)求。

20. 使用緩存優(yōu)化性能

頻繁請(qǐng)求API可能會(huì)導(dǎo)致性能問(wèn)題，可以使用緩存來(lái)優(yōu)化性能。我們可以使用requests_cache庫(kù)來(lái)實(shí)現(xiàn)緩存。

import requests_cache
import requests

# 啟用緩存
requests_cache.install_cache('api_cache', backend='sqlite', expire_after=3600)

# 發(fā)送請(qǐng)求
response = requests.get('https://api.example.com/data')

# 檢查請(qǐng)求是否成功
if response.status_code == 200:
    data = response.json()
    print(data)
else:
    print(f"請(qǐng)求失敗，狀態(tài)碼: {response.status_code}")

解釋：

requests_cache.install_cache啟用緩存，指定緩存后端和過(guò)期時(shí)間。
緩存會(huì)在第一次請(qǐng)求時(shí)存儲(chǔ)數(shù)據(jù)，后續(xù)請(qǐng)求會(huì)直接從緩存中讀取。

21. 使用OAuth2進(jìn)行認(rèn)證

許多現(xiàn)代API使用OAuth2進(jìn)行認(rèn)證。我們可以使用requests-oauthlib庫(kù)來(lái)處理OAuth2認(rèn)證。

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)

# 獲取訪問(wèn)令牌
token = oauth.fetch_token(token_url='https://api.example.com/oauth/token', client_id=client_id, client_secret=client_secret)

# 發(fā)送請(qǐng)求
response = oauth.get('https://api.example.com/data')

# 檢查請(qǐng)求是否成功
if response.status_code == 200:
    data = response.json()
    print(data)
else:
    print(f"請(qǐng)求失敗，狀態(tài)碼: {response.status_code}")

解釋：

BackendApplicationClient創(chuàng)建OAuth2客戶端。
oauth.fetch_token獲取訪問(wèn)令牌。
使用oauth.get發(fā)送帶有訪問(wèn)令牌的請(qǐng)求。

22. 處理API限流

許多API有請(qǐng)求頻率限制，我們需要學(xué)會(huì)如何處理這些限制。可以使用time.sleep來(lái)控制請(qǐng)求頻率。

import requests
import time

# 定義請(qǐng)求間隔時(shí)間
request_interval = 1  # 每秒最多發(fā)送一次請(qǐng)求

def fetch_data(page):
    response = requests.get(f'https://api.example.com/data?page={page}')
    return response.json()

# 獲取第一頁(yè)數(shù)據(jù)
data = fetch_data(1)

# 檢查是否有更多頁(yè)面
while 'next_page' in data and data['next_page']:
    next_page = data['next_page']
    time.sleep(request_interval)  # 控制請(qǐng)求頻率
    data = fetch_data(next_page)
    print(data)

解釋：

time.sleep(request_interval)控制每次請(qǐng)求之間的間隔時(shí)間。

23. 使用Web框架集成API

在Web應(yīng)用中，我們經(jīng)常需要集成API。可以使用Flask框架來(lái)實(shí)現(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)

解釋：