我正在测试一些dynamodb访问代码。过去，不正确的分页处理会导致错误(开发人员倾向于使用少量数据手动测试，因此很容易对分页的工作原理做出不正确的假设，这些假设只有在处理实际数据量后才会暴露出来(
我通常使用普通unittest和unittest.mock对访问代码进行单元测试，并以这种方式测试分页，但我最终编写了一些相当复杂的测试代码来模拟不同操作(扫描、查询batch_get_item(的分页。
我正在寻找一种更简单的方法来测试它; moto提供了一些希望。
但是，我真的不想将1MB +的数据加载到moto中以诱导分页，我想强制它对少量数据进行分页
所以我要问的症结是：
- moto 是否完全支持 DynamoDB 分页？
- 我可以配置分页阈值吗？
- 如何？
参考资料
- 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参数中对其进行配置。
分页具有以下核心概念：
hash_key+range_key_condition+filter_condition
- 要分页的 DynamoDB 记录列表
limit
- 查询返回的最大结果数
scan_index_forward
- 结果的顺序。您希望按range_key/sort_key按升序(例如 1、2、3(或降序(例如 3、2、1(对获取的记录进行排序
- >
last_evaluated_key
- 这指示数据库中上次处理的项目(对于密钥(。这会将该项目标记为将获取下一组项目的参考点。"无"表示从排序记录的开头开始查询。否则，请从指示的键开始查询。
- 把它想象成对
[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 =======================================================================================