我正在模擬一個回傳 dynamodb 查詢的內部函式。查詢begins_with拋出錯誤IndexError: list index out of range。
我更改了查詢并洗掉begins_with了但仍然遇到相同的錯誤。如果我洗掉AND condition from KeyConditionExpression,則查詢有效。
以下是查詢:
val = 'test#val#testing'
input_query = {
'TableName': <table_name>,
'KeyConditionExpression': '#23b62 = :23b62 And #23b63 = :23b63)',
'FilterExpression': 'contains(#23b64, :23b64)',
'ProjectionExpression': '#23b60,#23b61',
'ExpressionAttributeNames': {'#23b60': 'level', '#23b61': 'test_id', '#23b62': 'PK', '#23b63': 'SK', '#23b64': 'used_in'},
'ExpressionAttributeValues': {':23b62': {'S': 'testing'}, ':23b63': {'S': val}, ':23b64': {'S': 'test'}}
}
新查詢:
dynamodb_client.query(TableName="table",
KeyConditionExpression = "#PK = :PK And #SK = :SK",
ExpressionAttributeNames = {
"#PK": "PK",
"#SK": "SK"
},
FilterExpression = "contains(Used, :used)",
ExpressionAttributeValues ={
":PK": {"S": "tests"},
":SK": {"S": "test#en#testing"},
":used": {"S": "testing"}
}
)
測驗用例:
from botocore.exceptions import ClientError
from dynamodb_json import json_util as dynamodb_json
import logging
from contextlib import contextmanager
import pytest
from unittest.mock import patch
@contextmanager
def ddb_setup(dynamodb_resource):
table = dynamodb_resource.create_table(
TableName='table',
KeySchema=[
{
'AttributeName': 'PK',
'KeyType': 'HASH'
}, {
'AttributeName': 'SK',
'KeyType': 'SORT'
},
],
AttributeDefinitions=[
{
'AttributeName': 'PK',
'AttributeType': 'S'
}, {
'AttributeName': 'SK',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 1,
'WriteCapacityUnits': 1,
}
)
yield
class TestDynamoDB:
def test_create_table(self, dynamodb_resource, dynamodb_client):
with ddb_setup(dynamodb_resource):
try:
response = dynamodb_client.describe_table(
TableName='table')
resp = dynamodb_client.query(TableName="table",
KeyConditionExpression = "#PK = :PK And #SK = :SK",
ExpressionAttributeNames = {
"#PK": "PK",
"#SK": "SK"
},
FilterExpression = "contains(Used, :used)",
ExpressionAttributeValues ={
":PK": {"S": "tests"},
":SK": {"S": "test#en#testing"},
":used": {"S": "testing"}
}
)
except ClientError as err:
logger.error(f"error: {err.response['Error']['Code']}", )
assert err.response['Error']['Code'] == 'ResourceNotFoundException'
誰能建議我如何使用帶有 AND 條件的 moto 運行此查詢。
uj5u.com熱心網友回復:
這是一個使用 pytest 和 moto 的作業測驗配置示例。我添加了代碼來展示如何使用AND資源和客戶端 API 來使用條件。
import boto3
import boto3.dynamodb.conditions as conditions
import moto
import pytest
TABLE_NAME = "data"
@pytest.fixture
def test_table():
with moto.mock_dynamodb():
client = boto3.client("dynamodb")
client.create_table(
AttributeDefinitions=[
{"AttributeName": "PK", "AttributeType": "S"},
{"AttributeName": "SK", "AttributeType": "S"}
],
TableName=TABLE_NAME,
KeySchema=[
{"AttributeName": "PK", "KeyType": "HASH"},
{"AttributeName": "SK", "KeyType": "RANGE"}
],
BillingMode="PAY_PER_REQUEST"
)
table = boto3.resource("dynamodb").Table(TABLE_NAME)
table.put_item(Item={
"PK": "pk_value",
"SK": "sk_value"
})
yield TABLE_NAME
def test_query_with_and_using_resource(test_table):
table = boto3.resource("dynamodb").Table(TABLE_NAME)
response = table.query(
KeyConditionExpression=conditions.Key("PK").eq("pk_value") & conditions.Key("SK").eq("sk_value")
)
assert len(response["Items"]) == 1
def test_query_with_and_using_client(test_table):
client = boto3.client("dynamodb")
response = client.query(
TableName=TABLE_NAME,
KeyConditionExpression="#PK = :PK AND #SK = :SK",
ExpressionAttributeNames={
"#PK": "PK",
"#SK": "SK"
},
ExpressionAttributeValues={
":PK": {"S": "pk_value"},
":SK": {"S": "sk_value"}
}
)
assert len(response["Items"]) == 1
首先,我們用一個 dummy item 建立一個表,然后有兩個測驗,第一個測驗資源,第二個測驗客戶端 API。也許這可以幫助您找出錯誤。
uj5u.com熱心網友回復:
AWS 使用關鍵字RANGE來表示某事是排序鍵。(不知道為什么..)
如果您替換:
'KeyType': 'SORT'
測驗
'KeyType': 'RANGE'
通過。
我假設 AWS 在創建具有未知 KeyType 的表時會引發更明顯的錯誤。如果需要,您可以在 Moto 的 Github 上為 Moto 創建一個功能請求,以復制該行為并引發相同的例外。
轉載請註明出處,本文鏈接:https://www.uj5u.com/ruanti/477965.html
標籤:亚马逊网络服务 亚马逊-dynamodb 博托3 摩托
