IndexError: list index out of range with moto

IndexError: list index out of range with moto

我正在模拟一个返回 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 进行此查询。

这是使用 pytest 和 moto 的工作测试配置的示例。我添加了代码,展示了如何使用资源和客户端 API.

来使用 AND 条件
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

首先,我们设置一个带有虚拟项的 table,然后有两个测试,第一个用于资源,第二个用于客户端 API。也许这可以帮助您找出错误。

AWS 使用关键字 RANGE 来表示某物是 sort-key。 (不知道为什么..)

如果你更换: 'KeyType': 'SORT''KeyType': 'RANGE' 测试通过。


我假设 AWS 在创建具有未知 KeyType 的 table 时抛出更明显的错误。如果需要,您可以为 Moto 创建一个 feature request on Moto's Github 来复制该行为并抛出相同的异常。