使用 Python 和肉毒桿菌 3 對 Amazon DynamoDB 進行編程

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

本指南為想要將 Amazon DynamoDB 與 Python 搭配使用的程式設計師提供了一個方向。瞭解不同的抽象層、組態管理、錯誤處理、控制重試原則、管理保持活動狀態等。

關於博托

您可以使用適用於 Python 的官方 AWS 開發套件 (通常稱為 Boto3)，從 Python 存取 DynamoDB。Boto（發音為 boh-toh）這個名字來自 Amazon 河原產的淡水海豚。Boto3 圖書館是圖書館的第三個主要版本，於 2015 年首次發布。Boto3 程式庫相當龐大，因為它支援所有 AWS 服務，而不僅僅是 DynamoDB。此方向僅針對與 DynamoDB 相關的肉毒桿 3 部分。

Boto 由 AWS 作為託管的開源項目進行維護和發布。 GitHub它被分成兩個軟件包：肉毒核和 Boto3。

由於這些專案是在託管 GitHub，因此您可以檢視原始程式碼、追蹤未解決的問題，或提交您自己的問題。

使用博托文檔

使用下列資源開始使用 Boto 文件：

了解客戶端和資源抽象層

您將使用的兩個介面是用戶端介面和資源介面。

以下是使用用戶端介面插入項目的範例。請注意如何將所有值作為映射傳遞，其鍵指示其類型（'S' 表示字符串，'N' 表示數字）及其值作為字符串傳遞。這就是所 DynamoDB JSON 格式。

import boto3

dynamodb = boto3.client('dynamodb')

dynamodb.put_item(
    TableName='YourTableName',
    Item={
        'pk': {'S': 'id#1'},
        'sk': {'S': 'cart#123'},
        'name': {'S': 'SomeName'},
        'inventory': {'N': '500'},
        # ... more attributes ...
    }
)

這是使用資源接口的相同PutItem操作。數據類型是隱含的：

import boto3

dynamodb = boto3.resource('dynamodb')

table = dynamodb.Table('YourTableName')

table.put_item(
    Item={
        'pk': 'id#1',
        'sk': 'cart#123',
        'name': 'SomeName',
        'inventory': 500,
        # ... more attributes ...
    }
)

如果需要，您可以使用 boto3 提供的和TypeDeserializer類別，在一般 JSON TypeSerializer 和 DynamoDB JSON 之間進行轉換：

def dynamo_to_python(dynamo_object: dict) -> dict:
    deserializer = TypeDeserializer()
    return {
        k: deserializer.deserialize(v) 
        for k, v in dynamo_object.items()
    }  
  
def python_to_dynamo(python_object: dict) -> dict:
    serializer = TypeSerializer()
    return {
        k: serializer.serialize(v)
        for k, v in python_object.items()
    }

以下是如何使用客戶端界面執行查詢。它將查詢表示為 JSON 結構。它使用需要變量替換來處理任何潛在的關鍵KeyConditionExpression字衝突的字符串：

import boto3

client = boto3.client('dynamodb')

# Construct the query
response = client.query(
    TableName='YourTableName',
    KeyConditionExpression='pk = :pk_val AND begins_with(sk, :sk_val)',
    FilterExpression='#name = :name_val',
    ExpressionAttributeValues={
        ':pk_val': {'S': 'id#1'},
        ':sk_val': {'S': 'cart#'},
        ':name_val': {'S': 'SomeName'},
    },
    ExpressionAttributeNames={
        '#name': 'name',
    }
)

使用資源接口的相同查詢操作可以縮短和簡化：

import boto3
from boto3.dynamodb.conditions import Key, Attr

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('YourTableName')

response = table.query(
    KeyConditionExpression=Key('pk').eq('id#1') & Key('sk').begins_with('cart#'),
    FilterExpression=Attr('name').eq('SomeName')
)

作為最後一個例子，假設你想要獲得一個表的近似大小（這是保存在表上的元數據，大約每 6 小時更新一次）。使用客戶端界面，您可以執行describe_table()操作並從返回的 JSON 結構中提取答案：

import boto3

dynamodb = boto3.client('dynamodb')

response = dynamodb.describe_table(TableName='YourTableName')
size = response['Table']['TableSizeBytes']

透過資源介面，表格會隱含地執行描述作業，並直接將資料顯示為屬性：

import boto3

dynamodb = boto3.resource('dynamodb')

table = dynamodb.Table('YourTableName')
size = table.table_size_bytes

使用表格資源批次寫入器

只有更高層級的表格資源才能使用的便利之一是. batch_writer DynamoDB 支援批次寫入操作，允許在一個網路請求中執行多達 25 個置入或刪除操作。像這樣的批處理通過最大限度地減少網絡往返提高效率。

使用低階用戶端程式庫時，您可以使用此client.batch_write_item()作業執行批次。您必須手動將工作分割成 25 個批次。每次操作之後，您還必須請求接收未處理項目的列表（某些寫操作可能成功，而其他可能會失敗）。然後，您必須再次將這些未處理的項目傳遞到稍後的batch_write_item()操作中。沒有大量的樣板代碼。

該表格 .batch_Writer 方法創建一個前後關聯管理器，用於在批處理中寫入對象。它提供了一個界面，看起來好像你一次寫一個項目，但在內部它正在緩衝並批量發送項目。它還隱含處理未處理的項目重試。

dynamodb = boto3.resource('dynamodb')

table = dynamodb.Table('YourTableName')

movies = # long list of movies in {'pk': 'val', 'sk': 'val', etc} format
with table.batch_writer() as writer:
    for movie in movies:
        writer.put_item(Item=movie)

探索客戶端和資源層的其他代碼示例

您也可以參考下列程式碼範例儲存庫，以探索使用用戶端和資源的各種函式的使用方式：

了解用戶端和資源物件如何與工作階段和執行緒互動

Resource 對象不是線程安全的，不應該跨線程或進程共享。有關更多詳細信息，請參閱資源指南。

相比之下，Client 物件通常是執行緒安全的，除了特定的進階功能之外。如需詳細資訊，請參閱用戶端指南。

會話對象不是線程安全的。因此，每次在多線程環境中創建客戶端或資源時，都應首先創建一個新的會話，然後從會話中創建客戶端或資源。如需詳細資訊，請參閱工作階段指南。

當您呼叫時boto3.resource()，您隱含地使用預設工作階段。這對於編寫單線程代碼很方便。在編寫多線程代碼時，您需要首先為每個線程構建一個新的會話，然後從該會話中檢索資源：

# Explicitly create a new Session for this thread 
session = boto3.Session()
dynamodb = session.resource('dynamodb')

自定義 Config 對象

當構造一個客戶端或資源對象，你可以傳遞可選的命名參數來自定義行為。名為的參數可config解鎖各種功能。這是一個實例botocore.client.Config，Config 的參考文檔顯示了它公開供您控制的所有內容。配置指南提供了一個很好的概述。

Config 逾時

自定義配置的一種用途是調整網絡行為：

對於 DynamoDB 而言，60 秒的逾時時間過長。這意味著短暫的網絡故障將導致客戶端一分鐘的延遲，然後才能再試一次。下面的代碼將超時縮短到第二個：

import boto3
from botocore.config import Config

my_config = Config(
   connect_timeout = 1.0,
   read_timeout = 1.0
)
dynamodb = boto3.resource('dynamodb', config=my_config)

如需有關逾時的詳細討論，請參閱針對延遲感知 DynamoDB 應用程式調整 AWS Java SDK HTTP 要求設定。請注意，Java SDK 具有比 Python 更多的超時配置。

保持活著的 Config

如果您使用的是 botocore 1.27.84 或更高版本，您還可以控制 TCP 保持活動狀態：

將 TCP 保持活動狀態設置為True可以減少平均延遲。以下是示例代碼，當您擁有正確的 botocore 版本時，有條件地將 TCP 保持活動設置為 true：

import botocore
import boto3
from botocore.config import Config
from distutils.version import LooseVersion

required_version = "1.27.84"
current_version = botocore.__version__

my_config = Config(
   connect_timeout = 0.5,
   read_timeout = 0.5
)
if LooseVersion(current_version) > LooseVersion(required_version):
    my_config = my_config.merge(Config(tcp_keepalive = True))

dynamodb = boto3.resource('dynamodb', config=my_config)

閒置連接可以保持活躍的時間是有限的。如果您有閒置連接，但希望下一個請求使用已建立的連接，請考慮發送定期請求（例如每分鐘）。

重試 Config

該配置還接受一個名為 retries 的字典，您可以在其中指定所需的重試行為。當 SDK 收到錯誤且錯誤為暫時性類型時，會在 SDK 中重試。如果在內部重試錯誤（並且重試最終會產生成功的響應），則從調用代碼的角度看不到錯誤，只是稍微提高了延遲。以下是您可以指定的值：

您可以在重試指南以及 SDK 參考的「重試行為」主題中找到這些重試模式的擴充定義。

以下是一個明確使用legacy重試政策的示例，最多共有 3 個請求（2 次重試）：

import boto3
from botocore.config import Config

my_config = Config(
   connect_timeout = 1.0,
   read_timeout = 1.0,
   retries = {
     'mode': 'legacy',
     'total_max_attempts': 3
   }
)
dynamodb = boto3.resource('dynamodb', config=my_config)

由於 DynamoDB 是高可用性、低延遲的系統，因此您可能想要比內建重試原則允許的速度更積極。您可以通過將最大嘗試次數設置為 0，自己捕獲異常並根據自己的代碼重試，而不是依賴 boto3 進行隱式重試來實現自己的重試策略。

如果您管理自己的重試原則，則需要區分節流和錯誤：

最大池連接的 Config

最後，配置允許您控制連接池的大小：

此選項控制要保持集區以供重複使用的 HTTP 連線數目上限。每個工作階段會保留不同的池。如果您預期超過 10 個執行緒會針對用戶端或建立在相同工作階段的資源，您應該考慮提高這個問題，因此執行緒不必使用集區連線等待其他執行緒。

import boto3
from botocore.config import Config

my_config = Config(
   max_pool_connections = 20
)

# Setup a single session holding up to 20 pooled connections
session = boto3.Session(my_config)

# Create up to 20 resources against that session for handing to threads
# Notice the single-threaded access to the Session and each Resource
resource1 = session.resource('dynamodb')
resource2 = session.resource('dynamodb')
# etc

錯誤處理

AWS 服務異常並非所有在 Boto3 中靜態定義。這是因為 AWS 服務的錯誤和例外情況差異很大，並且可能會發生變化。Boto3 會將所有服務例外狀況包裝為 a，ClientError並將詳細資訊公開為結構化 JSON。例如，錯誤回應的結構可能如下：

{
    'Error': {
        'Code': 'SomeServiceException',
        'Message': 'Details/context around the exception or error'
    },
    'ResponseMetadata': {
        'RequestId': '1234567890ABCDEF',
        'HostId': 'host ID data will appear here as a hash',
        'HTTPStatusCode': 400,
        'HTTPHeaders': {'header metadata key/values will appear here'},
        'RetryAttempts': 0
    }
}

下列程式碼會擷取任何ClientError例外狀況，並查看Code中的字串值，以決定Error要採取的動作：

import botocore
import boto3

dynamodb = boto3.client('dynamodb')

try:
    response = dynamodb.put_item(...)

except botocore.exceptions.ClientError as err:
    print('Error Code: {}'.format(err.response['Error']['Code']))
    print('Error Message: {}'.format(err.response['Error']['Message']))
    print('Http Code: {}'.format(err.response['ResponseMetadata']['HTTPStatusCode']))
    print('Request ID: {}'.format(err.response['ResponseMetadata']['RequestId']))

    if err.response['Error']['Code'] in ('ProvisionedThroughputExceededException', 'ThrottlingException'):
        print("Received a throttle")
    elif err.response['Error']['Code'] == 'InternalServerError':
        print("Received a server error")
    else:
        raise err

某些（但不是全部）異常代碼已被實現為頂級類。您可以選擇直接處理這些問題。使用用戶端介面時，會在用戶端上動態填入這些例外狀況，並且您可以使用用戶端執行個體來 catch 取這些例外狀況，如下所示：

except ddb_client.exceptions.ProvisionedThroughputExceededException:

使用資源接口時，您必須使用從資源遍歷.meta.client到底層客戶端以訪問異常，如下所示：

except ddb_resource.meta.client.exceptions.ProvisionedThroughputExceededException:

若要複查具體化例外類型清單，您可以動態產生清單：

ddb = boto3.client("dynamodb")
print([e for e in dir(ddb.exceptions) if e.endswith('Exception') or e.endswith('Error')])

使用條件運算式執行寫入作業時，您可以要求如果運算式失敗，應在錯誤回應中傳回項目的值。

try:
    response = table.put_item(
        Item=item,
        ConditionExpression='attribute_not_exists(pk)',
        ReturnValuesOnConditionCheckFailure='ALL_OLD'
    )
except table.meta.client.exceptions.ConditionalCheckFailedException as e:
    print('Item already exists:', e.response['Item'])

有關錯誤處理和異常的進一步閱讀：

日誌

boto3 庫與 Python 的內置日誌記錄模塊集成在一起，用於跟踪會話期間發生的情況。若要控制記錄層級，您可以設定記錄模組：

import logging

logging.basicConfig(level=logging.INFO)

這將根記錄器配置為記錄INFO和高於級別的消息。記錄比級別不太嚴重的消息將被忽略。記錄層級包括DEBUGINFOWARNING、ERROR、、和CRITICAL。預設值為 WARNING。

在 boto3 記錄器是分層的。該庫使用了幾個不同的記錄器，每個記錄器都對應於庫的不同部分。您可以單獨控制每個行為：

其他程式庫也會記錄。在內部，boto3 使用第三方網址 3 進行 HTTP 連接處理。當延遲很重要時，您可以通過查看 urllib3 何時建立新連接或關閉閒置連接來確保池正常利用。

下列程式碼片段會INFO透過端點和連線集區活動的DEBUG記錄設定大部分記錄：

import logging

logging.getLogger('boto3').setLevel(logging.INFO)
logging.getLogger('botocore').setLevel(logging.INFO)
logging.getLogger('botocore.endpoint').setLevel(logging.DEBUG)
logging.getLogger('urllib3.connectionpool').setLevel(logging.DEBUG)

事件掛鉤

肉毒核在其執行的各個部分期間發出事件。您可以為這些事件註冊處理程序，以便無論何時發射事件，您的處理程序都會被調用。這使您可以擴展 botocore 的行為，而無需修改內部。

例如，假設您想要追蹤每次在應用程式中的任何 DynamoDB 表上呼叫PutItem作業時。您可以在'provide-client-params.dynamodb.PutItem'事件上註冊以 catch 和記錄每次在關聯的會話上調用PutItem操作時。範例如下：

import boto3
import botocore
import logging

def log_put_params(params, **kwargs):
    if 'TableName' in params and 'Item' in params:
        logging.info(f"PutItem on table {params['TableName']}: {params['Item']}")

logging.basicConfig(level=logging.INFO)

session = boto3.Session()
event_system = session.events

# Register our interest in hooking in when the parameters are provided to PutItem
event_system.register('provide-client-params.dynamodb.PutItem', log_put_params)

# Now, every time you use this session to put an item in DynamoDB,
# it will log the table name and item data.
dynamodb = session.resource('dynamodb')
table = dynamodb.Table('YourTableName')
table.put_item(
    Item={
        'pk': '123',
        'sk': 'cart#123',
        'item_data': 'YourItemData',
        # ... more attributes ...
    }
)

在處理程序中，您甚至可以通過編程方式操作參數以更改行為：

params['TableName'] = "NewTableName"

如需有關事件的詳細資訊，請參閱有關事件的 botocore 文件和事件的 boto3 文件。

分頁和分頁器

某些要求 (例如「查詢和掃描」) 會限制單一要求傳回的資料大小，並要求您重複提出要求以提取後續頁面。

您可以使用參數控制每個頁面要讀取的最大項目limit數。例如，如果您想要最後 10 個項目，則可以使用僅擷limit取最後 10 個項目。請注意，限制是在套用任何篩選之前，應從表格中讀取多少。沒有辦法在過濾後指定你想要的 10; 你只能控制預先過濾的計數，並在你實際檢索到 10 時檢查客戶端。無論限制如何，每個響應始終具有 1 MB 的最大大小。

如果響應包含 aLastEvaluatedKey，則表示響應因為達到計數或大小限制而結束。該鍵是為響應評估的最後一個鍵。您可以檢索它LastEvaluatedKey並將其傳遞給後續調用，以ExclusiveStartKey便從該起點讀取下一個塊。如果沒有LastEvaluatedKey返回，則表示沒有更多符合「查詢」或「掃描」的項目。

下面是一個簡單的例子（使用 Resource 接口，但客戶端接口具有相同的模式），每頁最多讀取 100 個項目並循環，直到讀取所有項目。

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('YourTableName')

query_params = {
    'KeyConditionExpression': Key('pk').eq('123') & Key('sk').gt(1000),
    'Limit': 100
}

while True:
    response = table.query(**query_params)

    # Process the items however you like
    for item in response['Items']:
        print(item)

    # No LastEvaluatedKey means no more items to retrieve
    if 'LastEvaluatedKey' not in response:
        break

    # If there are possibly more items, update the start key for the next page
    query_params['ExclusiveStartKey'] = response['LastEvaluatedKey']

為了方便起見，boto3 可以使用分頁器為您做到這一點。但是，它僅適用於客戶端界面。以下是重寫以使用分頁器的代碼：

import boto3

dynamodb = boto3.client('dynamodb')

paginator = dynamodb.get_paginator('query')

query_params = {
    'TableName': 'YourTableName',
    'KeyConditionExpression': 'pk = :pk_val AND sk > :sk_val',
    'ExpressionAttributeValues': {
        ':pk_val': {'S': '123'},
        ':sk_val': {'N': '1000'},
    },
    'Limit': 100
}

page_iterator = paginator.paginate(**query_params)

for page in page_iterator:
    # Process the items however you like
    for item in page['Items']:
        print(item)

如需詳細資訊，請參閱分頁器指南和動態分頁器的 API 參考。

等待程式

服務員提供了在繼續之前等待某些事情完成的能力。目前，它們僅支持等待創建或刪除表。在後台，服務員操作每 20 秒為您檢查一次，最多 25 次。您可以自己做，但是在編寫自動化時使用服務員很優雅。

此代碼顯示瞭如何等待特定表已創建：

# Create a table, wait until it exists, and print its ARN
response = client.create_table(...)
waiter = client.get_waiter('table_exists')
waiter.wait(TableName='YourTableName')
print('Table created:', response['TableDescription']['TableArn']

有關更多信息，請參閱服務員指南和服務員參考。