python - 如何使用 moto 模拟 DynamoDB 分页？

Question

我正在测试一些 dynamodb 访问代码。在过去，对分页的不正确处理导致了错误(开发人员倾向于使用少量数据进行手动测试，因此很容易对分页的工作原理做出错误的假设，只有在处理实际数据量时才会出现这种假设)

我通常使用普通 unittest 和 unittest.mock 对访问代码进行单元测试，并以这种方式测试分页，但我最终写了一些合理的复杂的测试代码来模拟不同操作的分页(扫描、查询、batch_get_item)。

我正在寻找一种更简单的测试方法； moto 带来了一些希望

但是，我真的不想将 1MB+ 的数据加载到 moto 中以引起分页，我想强制它对少量数据进行分页

所以我要问的关键是:

• moto 是否完全支持 DynamoDB 分页？
• 我可以配置分页阈值吗？
• 如何？

Answer 1

引用资料

• https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Query.Pagination.html
• https://pynamodb.readthedocs.io/en/latest/api.html

moto 是否完全支持 DynamoDB 分页？

是的，它通过 moto.mock_dynamodb2 功能实现。我尝试使用 PynamoDB 的 query 功能进行分页，它在 moto.mock_dynamodb2 提供的模拟 DynamoDB 环境中运行良好。

我可以配置分页阈值吗？

使用PynamoDB的query，可以在limit参数中配置。

分页具有以下核心概念:

1. hash_key + range_key_condition + filter_condition

• 要分页的 DynamoDB 记录列表

1. 限制

• 查询返回的最大结果数

1. scan_index_forward

• 结果的顺序。您希望按 range_key/sort_key 升序(例如 1、2、3)或降序(例如 3、2、1)对获取的记录进行排序

1. last_evaluated_key

• 这表示数据库中最后处理的项目(对于键)。这将该项目标记为将获取下一组项目的引用点。 None 表示从已排序记录的开头查询。否则，从指定的键开始查询。
• 将其想象成对 [0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50] 的二进制搜索。如果我们从头开始分页 4 个项目，我们将得到 [0, 5, 10, 15]。如果我们想要获得接下来的 4 个项目，我们不需要从开始(0)一直迭代到目标(20 起)。这种算法在最坏情况下会导致线性 O(n) 时间复杂度，其中 n 是所有记录的计数。相反，我们可以做的是对大于最后获取的项目(15)的第一个项目执行二进制搜索，我们将在对数中得到 20 O(log(n)).

怎么做？

查看 Python 代码片段

# Testing date: 2020 9September 29

# Versions
# moto==1.3.16
# pynamodb==4.3.3
# pytest==6.1.0

import itertools

from moto import mock_dynamodb2
from pynamodb.attributes import *
from pynamodb.models import Model
import pytest


# Model

class Location(Model):
    class Meta:
        table_name = 'Location-table'
        region = 'ap-southeast-1'

    continent = UnicodeAttribute(hash_key=True)  # also known as partition_key
    country = UnicodeAttribute(range_key=True)  # also known as sort_key
    capital = UnicodeAttribute()
    gmt = NumberAttribute()

    def __iter__(self):
        for name, attr in self.get_attributes().items():
            yield name, attr.serialize(getattr(self, name))


# Test data

LOCATIONS = [
    {
        'continent': 'Europe',
        'country': 'Spain',
        'capital': 'Madrid',
        'gmt': 2,
    },
    {
        'continent': 'Europe',
        'country': 'Germany',
        'capital': 'Berlin',
        'gmt': 2,
    },
    {
        'continent': 'South America',
        'country': 'Venezuela',
        'capital': 'Caracas',
        'gmt': -4,
    },
    {
        'continent': 'Europe',
        'country': 'Ukraine',
        'capital': 'Kyiv',
        'gmt': 3,
    },
    {
        'continent': 'South America',
        'country': 'Brazil',
        'capital': 'Brasília',
        'gmt': -3,
    },
    {
        'continent': 'Europe',
        'country': 'Finland',
        'capital': 'Helsinki',
        'gmt': 3,
    },
    {
        'continent': 'Europe',
        'country': 'Ireland',
        'capital': 'Dublin',
        'gmt': 1,
    },
]


# Test algorithms

def _setup_table(locations):
    Location.create_table()

    for location in locations:
        Location(**location).save()

def _get_filter_condition():
    # Put logic here for the filter condition. Uncomment the code below to try.
    # filter_condition = (Location.gmt >= 2) \
    #                     & (Location.capital.contains('in') | Location.capital.startswith('A'))
    # return filter_condition
    return None


@mock_dynamodb2
def test_dynamodb_pagination():
    _setup_table(LOCATIONS)
    filter_condition = _get_filter_condition()

    # Expected query order for Europe. This should be sorted by country (which is the sort_key field).
    SORTED_EUROPE_COUNTRIES = [
        'Finland',
        'Germany',
        'Ireland',
        'Spain',
        'Ukraine',
    ]
    country_index = 0

    # This indicates the last processed item (for the key) from the database. This marks that item
    # as the reference point to where the next set of items will be fetched. None means query from
    # the beginning of the sorted records. Otherwise, start the query from the indicated key.
    last_evaluated_key = None

    for query_index in itertools.count(0):
        result = Location.query(
            hash_key='Europe',
            filter_condition=filter_condition,  # Filter the query results
            limit=2,  # Maximum number of items to fetch from the database
            last_evaluated_key=last_evaluated_key,  # The reference starting point of the fetch
            scan_index_forward=True,  # Indicate if in lexicographical order (increasing) or in reverse (decreasing)
        )

        for item in result:
            print(f"Query #{query_index} - Country #{country_index} - {item}")

            assert item.country == SORTED_EUROPE_COUNTRIES[country_index]
            country_index += 1

        print(f"result.last_evaluated_key {result.last_evaluated_key}\n")
        last_evaluated_key = result.last_evaluated_key

        if last_evaluated_key is None:
            print(f"Reached the last queried item in the database")
            break

输出:

(venv) nponcian 2020_9Sep_10_DynamoDB$ pytest pagination_test.py -rP
====================================================================================== test session starts ======================================================================================
platform linux -- Python 3.8.2, pytest-6.1.0, py-1.9.0, pluggy-0.13.1
rootdir: /home/nponcian/Documents/Program/2020_9Sep_10_DynamoDB
plugins: cov-2.10.1, mock-3.3.1
collected 1 item                                                                                                                                                                                

pagination_test.py .                                                                                                                                                                      [100%]

============================================================================================ PASSES =============================================================================================
___________________________________________________________________________________ test_dynamodb_pagination ____________________________________________________________________________________
------------------------------------------------------------------------------------- Captured stdout call --------------------------------------------------------------------------------------
Query #0 - Country #0 - Location-table<Europe, Finland>
Query #0 - Country #1 - Location-table<Europe, Germany>
result.last_evaluated_key {'continent': {'S': 'Europe'}, 'country': {'S': 'Germany'}}

Query #1 - Country #2 - Location-table<Europe, Ireland>
Query #1 - Country #3 - Location-table<Europe, Spain>
result.last_evaluated_key {'continent': {'S': 'Europe'}, 'country': {'S': 'Spain'}}

Query #2 - Country #4 - Location-table<Europe, Ukraine>
result.last_evaluated_key None

Reached the last queried item in the database
======================================================================================= 1 passed in 0.40s =======================================================================================