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

使用 Python 和 Boto3 程式設計 Amazon DynamoDB

本指南提供程式設計人員想要搭配 Python 使用 Amazon DynamoDB 的方向。了解不同的抽象層、組態管理、錯誤處理、控制重試政策、管理保持連線等。

關於 Boto

您可以使用適用於 Python 的官方 AWS SDK，從 Python 存取 DynamoDB，通常稱為 Boto3。Boto （發音 boh-toh) 的名稱來自 Amazon River 原生的淡水海豚。Boto3 程式庫是程式庫的第三個主要版本，於 2015 年首次發行。Boto3 程式庫相當大，因為它支援所有 AWS 服務，而不只是 DynamoDB。此方向僅針對與 DynamoDB 相關的 Boto3 部分。

Boto 由 維護和發佈 AWS ，做為託管在 GitHub 上的開放原始碼專案。它分為兩個套件：Botocore 和 Boto3。

由於這些專案託管在 GitHub 上，因此您可以檢視原始碼、追蹤開啟的問題，或提交您自己的問題。

使用 Boto 文件

使用下列資源開始使用 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

一個只有更高層級的資料表資源才可用的便利性是 batch_writer。DynamoDB 支援批次寫入操作，可在一個網路請求中最多放置或刪除 25 個操作。像這樣的批次處理可最大限度地減少網路往返，進而提高效率。

使用低階用戶端程式庫時，您可以使用 client.batch_write_item()操作來執行批次。您必須手動將工作分割為 25 個批次。在每次操作後，您也必須請求 接收未處理項目的清單 （有些寫入操作可能會成功，而其他操作可能會失敗）。然後，您必須將這些未處理的項目再次傳遞至稍後batch_write_item()的操作。有大量的樣板程式碼。

Table.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)

探索用戶端和資源層的其他程式碼範例

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

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

資源物件不安全，不應跨執行緒或程序共用。如需詳細資訊，請參閱 資源指南。

相反地，用戶端物件通常可以安全執行緒，但特定進階功能除外。如需詳細資訊，請參閱 用戶端的指南。

工作階段物件執行緒不安全。因此，每次在多執行緒環境中建立用戶端或資源時，您應先建立新的工作階段，然後從工作階段建立用戶端或資源。如需詳細資訊，請參閱工作階段指南。

當您呼叫 時boto3.resource()，您會隱含地使用預設工作階段。這對於編寫單執行緒程式碼非常方便。撰寫多執行緒程式碼時，您會想要先為每個執行緒建構新的工作階段，然後從該工作階段擷取資源：

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

自訂 Config 物件

建構用戶端或資源物件時，您可以傳遞選用的具名參數來自訂行為。名為 的 參數會config解除鎖定各種功能。這是 的執行個體，botocore.client.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 開發套件的逾時組態比 Python 多。

保持連線的組態

如果您使用的是 botocore 1.27.84 或更新版本，您也可以控制 TCP Keep-Alive：

將 TCP Keep-Alive 設定為 True可減少平均延遲。以下是範例程式碼，當您有正確的 botocore 版本時，將 TCP Keep-Alive 設定為 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)

閒置連線可以保持運作的時間有限制。如果您有閒置連線，但希望下一個請求使用已建立的連線，請考慮定期傳送請求 （每分鐘傳送一次）。

重試的組態

此組態也接受名為重試的字典，您可以在其中指定所需的重試行為。當 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 進行隱含重試，來實作您自己的重試政策。

如果您管理自己的重試政策，建議您區分調節和錯誤：

集區連線上限的組態

最後，組態可讓您控制連線集區大小：

此選項可控制 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 將所有服務例外狀況包裝為 ，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

有些 （但不是全部） 例外狀況代碼已具體化為最上層類別。您可以選擇直接處理這些項目。使用用戶端界面時，這些例外狀況會在您的用戶端上動態填入，而您使用用戶端執行個體來擷取這些例外狀況，如下所示：

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 和更高層級的訊息。系統會忽略較層級不嚴重的記錄訊息。記錄層級包括 DEBUG、INFO、ERROR、 WARNING和 CRITICAL。預設值為 WARNING。

boto3 中的記錄器是階層式的。程式庫使用幾個不同的記錄器，每個對應至程式庫的不同部分。您可以個別控制每個 的行為：

其他程式庫日誌也一樣。在內部，boto3 會使用第三方 urllib3 來處理 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 在執行的各個部分期間發出事件。您可以為這些事件註冊處理常式，以便每次發出事件時，都會呼叫您的處理常式。這可讓您擴充 botocore 的行為，而不必修改內部。

例如，假設您想要追蹤每次在應用程式中任何 DynamoDB 資料表上呼叫PutItem操作時。您可以在'provide-client-params.dynamodb.PutItem'事件上註冊，以在每次在關聯的工作階段上叫用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。

如果回應包含 LastEvaluatedKey，則表示回應因為達到計數或大小限制而結束。金鑰是針對回應評估的最後一個金鑰。您可以擷取此項目，LastEvaluatedKey並將其傳遞至後續呼叫ExclusiveStartKey，以讀取該起點的下一個區塊。沒有LastEvaluatedKey傳回該項目時， 表示沒有更多符合查詢或掃描的項目。

以下是一個簡單的範例 （使用 資源界面，但用戶端界面具有相同的模式），每個頁面和迴圈最多讀取 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 可以使用 Paginator 為您執行此操作。不過，它僅適用於用戶端界面。以下是改寫為使用 Paginator 的程式碼：

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)

如需詳細資訊，請參閱 Paginator 指南和 DynamoDB.Paginator.Query 的 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']

如需詳細資訊，請參閱等待者指南和等待者參考。