I am mocking an internal function which is returning dynamodb query. the query had begins_with which was throwing error IndexError: list index out of range.

I changed the query and removed begins_with yet still getting the same error. If I remove AND condition from KeyConditionExpression then the query works.

Below is the query:

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'}}
}

New Query :

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"}
            }
            )

Test case:

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'

Could anyone suggest how can I run this query with moto with AND condition.

CodePudding user response：

Here is an example of a working test configuration using pytest and moto. I've added code that shows how to use the AND condition using the resource and client 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

First, we set up a table with a dummy item, and then there are two tests, the first for the resource and the second for the client API. Maybe this helps you figure out the mistake.

CodePudding user response：

AWS uses the keyword RANGE to indicate that something is a sort-key. (No idea why..)

If you replace: 'KeyType': 'SORT' with 'KeyType': 'RANGE' the test passes.

I'm assuming that AWS throws a more obvious error when creating a table with an unknown KeyType. If you want, you can create a feature request on Moto's Github for Moto to replicate that behaviour and throw the same exception.