안녕하세요. Cursor Pagination Pattern(또는 Cursor Paging Pattern)의 예를 공유하고 싶었습니다. 검색할 때 앞으로 이동하는 경우의 예만 찾을 수 있었지만 뒤로도 안되고, 시작과 끝에서 데이터를 처리하는 방법도 없습니다.
여기에서 이에 대한 저장소를 볼 수 있지만 여기에서 모든 것을 설명하겠습니다.
저는 패키지 관리 도구로 Python Poetry를 사용하므로 이 예에서는 여러분이 이미 Python Poetry를 가지고 있다고 가정합니다. 가장 먼저 해야 할 일은 시 설치로 종속성을 설치하는 것입니다. pip를 사용하여 pip install pymongo loguru를 사용해 설치할 수도 있습니다.
이제 Mongo 데이터베이스도 필요합니다. 여기에서 MongoDB Community Edition을 다운로드하고 이 가이드를 사용하여 구성할 수 있습니다.
이제 종속성이 설치되었고 데이터베이스가 설치되었으므로 여기에 데이터를 추가할 수 있습니다. 이를 위해 다음을 사용할 수 있습니다.
from pymongo import MongoClient
# Data to add
sample_posts = [
{"title": "Post 1", "content": "Content 1", "date": datetime(2023, 8, 1)},
{"title": "Post 2", "content": "Content 2", "date": datetime(2023, 8, 2)},
{"title": "Post 3", "content": "Content 3", "date": datetime(2023, 8, 3)},
{"title": "Post 4", "content": "Content 4", "date": datetime(2023, 8, 4)},
{"title": "Post 5", "content": "Content 5", "date": datetime(2023, 8, 5)},
{"title": "Post 6", "content": "Content 6", "date": datetime(2023, 8, 6)},
{"title": "Post 7", "content": "Content 7", "date": datetime(2023, 8, 7)},
{"title": "Post 8", "content": "Content 8", "date": datetime(2023, 8, 8)},
{"title": "Post 9", "content": "Content 9", "date": datetime(2023, 8, 9)},
{"title": "Post 10", "content": "Content 10", "date": datetime(2023, 8, 10)},
{"title": "Post 11", "content": "Content 11", "date": datetime(2023, 8, 11)},
]
# Creating connection
token = "mongodb://localhost:27017"
client = MongoClient(token)
cursor_db = client.cursor_db.content
cursor_db.insert_many(sample_posts)
이를 통해 컬렉션 콘텐츠에 대한 로컬 데이터베이스 연결을 만듭니다. 그런 다음 sample_posts의 값을 여기에 추가합니다. 이제 검색할 데이터가 있으므로 쿼리를 시작할 수 있습니다. 끝까지 데이터를 검색하고 읽어봅시다.
# Import libraries
from bson.objectid import ObjectId
from datetime import datetime
from loguru import logger
from pymongo import MongoClient
# Use token to connect to local database
token = "mongodb://localhost:27017"
client = MongoClient(token)
# Access cursor_db collection (it will be created if it does not exist)
cursor_db = client.cursor_db.content
default_page_size = 5
def fetch_next_page(cursor, page_size = None):
# Use the provided page_size or use a default value
page_size = page_size or default_page_size
# Check if there is a cursor
if cursor:
# Get documents with `_id` greater than the cursor
query = {"_id": {'$gt': cursor}}
else:
# Get everything
query = {}
# Sort in ascending order by `_id`
sort_order = 1
# Define the aggregation pipeline
pipeline = [
{"$match": query}, # Filter based on the cursor
{"$sort": {"_id": sort_order}}, # Sort documents by `_id`
{"$limit": page_size + 1}, # Limit results to page_size + 1 to check if there's a next page
# {"$project": {"_id": 1, "title": 1, "content": 1}} # In case you want to return only certain attributes
]
# Execute the aggregation pipeline
results = list(cursor_db.aggregate(pipeline))
# logger.debug(results)
# Validate if some data was found
if not results: raise ValueError("No data found")
# Check if there are more documents than the page size
if len(results) > page_size:
# Deleting extra document
results.pop(-1)
# Set the cursor for the next page
next_cursor = results[-1]['_id']
# Set the previous cursor
if cursor:
# in case the cursor have data
prev_cursor = results[0]['_id']
else:
# In case the cursor don't have data (first page)
prev_cursor = None
# Indicate you haven't reached the end of the data
at_end = False
else:
# Indicate that there are not more pages available (last page reached)
next_cursor = None
# Set the cursor for the previous page
prev_cursor = results[0]['_id']
# Indicate you have reached the end of the data
at_end = True
return results, next_cursor, prev_cursor, at_end
@logger.catch
def main():
"""Main function."""
# Get the first page
results, next_cursor, prev_cursor, at_end = fetch_next_page(None)
logger.info(f"{results = }")
logger.info(f"{next_cursor = }")
logger.info(f"{prev_cursor = }")
logger.info(f"{at_end = }")
if __name__:
main()
logger.info("--- Execution end ---")
해당 코드는 다음을 반환합니다.
2024-09-02 08:55:24.388 | INFO | __main__:main:73 - results = [{'_id': ObjectId('66bdfdcf7a0667fd1888c20c'), 'title': 'Post 1', 'content': 'Content 1', 'date': datetime.datetime(2023, 8, 1, 0, 0)}, {'_id': ObjectId('66bdfdcf7a0667fd1888c20d'), 'title': 'Post 2', 'content': 'Content 2', 'date': datetime.datetime(2023, 8, 2, 0, 0)}, {'_id': ObjectId('66bdfdcf7a0667fd1888c20e'), 'title': 'Post 3', 'content': 'Content 3', 'date': datetime.datetime(2023, 8, 3, 0, 0)}, {'_id': ObjectId('66bdfdcf7a0667fd1888c20f'), 'title': 'Post 4', 'content': 'Content 4', 'date': datetime.datetime(2023, 8, 4, 0, 0)}, {'_id': ObjectId('66bdfdcf7a0667fd1888c210'), 'title': 'Post 5', 'content': 'Content 5', 'date': datetime.datetime(2023, 8, 5, 0, 0)}]
2024-09-02 08:55:24.388 | INFO | __main__:main:74 - next_cursor = ObjectId('66bdfdcf7a0667fd1888c210')
2024-09-02 08:55:24.388 | INFO | __main__:main:75 - prev_cursor = None
2024-09-02 08:55:24.388 | INFO | __main__:main:76 - at_end = False
2024-09-02 08:55:24.388 | INFO | __main__:<module>:79 - --- Execution end ---
커서가 다음 페이지를 가리키고 이전 페이지가 없음인 것을 볼 수 있으며, 데이터의 끝이 아님을 식별합니다. 이 값을 얻으려면 fetch_next_page 함수를 자세히 살펴봐야 합니다. 여기서는 page_size, 쿼리, sort_order를 정의한 다음 집계 작업에 대한 파이프라인을 생성한 것을 볼 수 있습니다. 다른 정보 페이지가 있는지 식별하기 위해 $limit 연산자를 사용하고 page_size + 1 값을 제공하여 실제로 해당 + 1이 있는 다른 페이지가 있는지 확인합니다. 실제로 이를 확인하려면 len( 결과) > page_size, 반환된 데이터 수가 page_size보다 큰 경우 다른 페이지가 있습니다. 오히려 이게 마지막 페이지입니다.
다음 페이지가 있는 경우 쿼리한 정보 목록에서 마지막 요소를 삭제해야 합니다. 이는 파이프라인의 +1이었기 때문에 현재 마지막 값의 _id로 next_cursor를 설정해야 합니다. 목록을 작성하고 경우에 따라 prev_cursor(이전 커서)를 설정합니다. 이 커서 앞에 데이터가 있다는 의미의 커서가 있으면 다른 경우에는 이것이 첫 번째 데이터 그룹이라는 의미이므로 다음이 있습니다. 이전 정보가 없으므로 커서는 발견된 데이터의 첫 번째 _id이거나 None이어야 합니다.
이제 데이터를 검색하고 몇 가지 중요한 유효성 검사를 추가하는 방법을 알았으므로 데이터를 앞으로 탐색하는 방법을 활성화해야 합니다. 이를 위해 입력 명령을 사용하여 스크립트를 실행하는 사용자에게 이동 방향을 작성하도록 요청합니다. 하지만 지금은 정방향(f)만 가능합니다. 다음과 같이 기본 기능을 업데이트할 수 있습니다.
@logger.catch
def main():
"""Main function."""
# Get the first page
results, next_cursor, prev_cursor, at_end = fetch_next_page(None)
logger.info(f"{results = }")
logger.info(f"{next_cursor = }")
logger.info(f"{prev_cursor = }")
logger.info(f"{at_end = }")
# Checking if there is more data to show
if next_cursor:
# Enter a cycle to traverse the data
while(True):
print(125 * "*")
# Ask for the user to move forward or cancel the execution
inn = input("Can only move Forward (f) or Cancel (c): ")
# Execute action acording to the input
if inn == "f":
results, next_cursor, prev_cursor, at_end = fetch_next_page(next_cursor, default_page_size)
elif inn == "c":
logger.warning("------- Canceling execution -------")
break
else:
# In case the user sends something that is not a valid option
print("Not valid action, it can only move in the opposite direction.")
continue
logger.info(f"{results = }")
logger.info(f"{next_cursor = }")
logger.info(f"{prev_cursor = }")
logger.info(f"{at_end = }")
else:
logger.warning("There is not more data to show")
이를 통해 데이터를 끝까지 탐색할 수 있지만 끝에 도달하면 처음으로 돌아가고 주기가 다시 시작됩니다. 따라서 이를 피하고 뒤로 이동하려면 몇 가지 검증을 추가해야 합니다. 이를 위해 fetch_previous_page 함수를 만들고 main 함수에 몇 가지 변경 사항을 추가하겠습니다.
def fetch_previous_page(cursor, page_size = None):
# Use the provided page_size or fallback to the class attribute
page_size = page_size or default_page_size
# Check if there is a cursor
if cursor:
# Get documents with `_id` less than the cursor
query = {'_id': {'$lt': cursor}}
else:
# Get everything
query = {}
# Sort in descending order by `_id`
sort_order = -1
# Define the aggregation pipeline
pipeline = [
{"$match": query}, # Filter based on the cursor
{"$sort": {"_id": sort_order}}, # Sort documents by `_id`
{"$limit": page_size + 1}, # Limit results to page_size + 1 to check if there's a next page
# {"$project": {"_id": 1, "title": 1, "content": 1}} # In case you want to return only certain attributes
]
# Execute the aggregation pipeline
results = list(cursor_db.aggregate(pipeline))
# Validate if some data was found
if not results: raise ValueError("No data found")
# Check if there are more documents than the page size
if len(results) > page_size:
# Deleting extra document
results.pop(-1)
# Reverse the results to maintain the correct order
results.reverse()
# Set the cursor for the previous page
prev_cursor = results[0]['_id']
# Set the cursor for the next page
next_cursor = results[-1]['_id']
# Indicate you are not at the start of the data
at_start = False
else:
# Reverse the results to maintain the correct order
results.reverse()
# Indicate that there are not more previous pages available (initial page reached)
prev_cursor = None
# !!!!
next_cursor = results[-1]['_id']
# Indicate you have reached the start of the data
at_start = True
return results, next_cursor, prev_cursor, at_start
fetch_next_page와 매우 유사하지만 쿼리(조건이 충족되는 경우)는 $lt 연산자를 사용하고 sort_order는 -1이어야 데이터를 필요한 순서로 가져옵니다. 이제 len(results) > page_size, 조건이 true인 경우 추가 요소를 제거하고 데이터가 올바르게 표시되도록 데이터 순서를 바꾼 다음 이전 커서를 데이터의 첫 번째 요소로 설정하고 다음 커서를 설정합니다. 마지막까지. 반대로 데이터가 반전되어 이전 커서는 None으로 설정되고(이전 데이터가 없기 때문에) 다음 커서는 목록의 마지막 값으로 설정됩니다. 두 경우 모두 이 상황을 식별하기 위해 at_start라는 부울 변수가 정의됩니다. 이제 주 기능에서 뒤로 이동하기 위한 사용자와의 상호 작용을 추가해야 하므로 데이터의 시작, 끝 또는 중간에 있는 경우 처리해야 할 3가지 상황이 있습니다. 앞으로만 진행, 뒤로만 진행 , 앞으로 또는 뒤로 이동:
@logger.catch
def main():
"""Main function."""
# Get the first page
results, next_cursor, prev_cursor, at_end = fetch_next_page(None)
logger.info(f"{results = }")
logger.info(f"{next_cursor = }")
logger.info(f"{prev_cursor = }")
logger.info(f"{at_end = }")
# Checking if there is more data to show
if not(at_start and at_end):
# Enter a cycle to traverse the data
while(True):
print(125 * "*")
# Ask for the user to move forward or cancel the execution
if at_end:
inn = input("Can only move Backward (b) or Cancel (c): ")
stage = 0
elif at_start:
inn = input("Can only move Forward (f) or Cancel (c): ")
stage = 1
else:
inn = input("Can move Forward (f), Backward (b), or Cancel (c): ")
stage = 2
# Execute action acording to the input
if inn == "f" and stage in [1, 2]:
results, next_cursor, prev_cursor, at_end = fetch_next_page(next_cursor, page_size)
# For this example, you must reset here the value, otherwise you lose the reference of the cursor
at_start = False
elif inn == "b" and stage in [0, 2]:
results, next_cursor, prev_cursor, at_start = fetch_previous_page(prev_cursor, page_size)
# For this example, you must reset here the value, otherwise you lose the reference of the cursor
at_end = False
elif inn == "c":
logger.warning("------- Canceling execution -------")
break
else:
print("Not valid action, it can only move in the opposite direction.")
continue
logger.info(f"{results = }")
logger.info(f"{next_cursor = }")
logger.info(f"{prev_cursor = }")
logger.info(f"{at_start = }")
logger.info(f"{at_end = }")
else:
logger.warning("There is not more data to show")
데이터를 탐색하는 동안 현재 있는 단계를 식별하기 위해 사용자 입력에 유효성 검사를 추가했습니다. 또한 fetch_next_page 및 fetch_previous_page 실행 후 각각 at_start 및 at_end에 도달한 후 재설정해야 합니다. 제한. 이제 데이터 끝에 도달하고 시작될 때까지 뒤로 이동할 수 있습니다. 데이터의 첫 번째 페이지를 가져온 후 유효성 검사가 업데이트되어 at_start 및 at_end 플래그가 True인지 확인하여 더 이상 표시할 데이터가 없음을 나타냅니다.
Note: I was facing a bug at this point which I cannot reproduce right now, but it was causing problems when going backward and reaching the start, the cursor was pointing to the wrong place and when you wanted to go forward it skip 1 element. To solve it I added a validation in fetch_previous_page if a parameter called prev_at_start (which is the previous value of at_start) to assing next_cursor the value results[0]['_id'] or, results[-1]['_id'] in case the previous stage was not at the beginning of the data. This will be ommited from now on, but I think is worth the mention.
Now that we can traverse the data from beginning to end and going forward or backward in it, we can create a class that have all this functions and call it to use the example. Also we must add the docstring so everything is documents correctly. The result of that are in this code:
"""Cursor Paging/Pagination Pattern Example."""
from bson.objectid import ObjectId
from datetime import datetime
from loguru import logger
from pymongo import MongoClient
class cursorPattern:
"""
A class to handle cursor-based pagination for MongoDB collections.
Attributes:
-----------
cursor_db : pymongo.collection.Collection
The MongoDB collection used for pagination.
page_size : int
Size of the pages.
"""
def __init__(self, page_size: int = 5) -> None:
"""Initializes the class.
Sets up a connection to MongoDB and specifying
the collection to work with.
"""
token = "mongodb://localhost:27017"
client = MongoClient(token)
self.cursor_db = client.cursor_db.content
self.page_size = page_size
def add_data(self,) -> None:
"""Inserts sample data into the MongoDB collection for demonstration purposes.
Note:
-----
It should only use once, otherwise you will have repeated data.
"""
sample_posts = [
{"title": "Post 1", "content": "Content 1", "date": datetime(2023, 8, 1)},
{"title": "Post 2", "content": "Content 2", "date": datetime(2023, 8, 2)},
{"title": "Post 3", "content": "Content 3", "date": datetime(2023, 8, 3)},
{"title": "Post 4", "content": "Content 4", "date": datetime(2023, 8, 4)},
{"title": "Post 5", "content": "Content 5", "date": datetime(2023, 8, 5)},
{"title": "Post 6", "content": "Content 6", "date": datetime(2023, 8, 6)},
{"title": "Post 7", "content": "Content 7", "date": datetime(2023, 8, 7)},
{"title": "Post 8", "content": "Content 8", "date": datetime(2023, 8, 8)},
{"title": "Post 9", "content": "Content 9", "date": datetime(2023, 8, 9)},
{"title": "Post 10", "content": "Content 10", "date": datetime(2023, 8, 10)},
{"title": "Post 11", "content": "Content 11", "date": datetime(2023, 8, 11)},
]
self.cursor_db.insert_many(sample_posts)
def _fetch_next_page(
self, cursor: ObjectId | None, page_size: int | None = None
) -> tuple[list, ObjectId | None, ObjectId | None, bool]:
"""Retrieves the next page of data based on the provided cursor.
Parameters:
-----------
cursor : ObjectId | None
The current cursor indicating the last document of the previous page.
page_size : int | None
The number of documents to retrieve per page (default is the class's page_size).
Returns:
--------
tuple:
- results (list): The list of documents retrieved.
- next_cursor (ObjectId | None): The cursor pointing to the start of the next page, None in case is the last page.
- prev_cursor (ObjectId | None): The cursor pointing to the start of the previous page, None in case is the start page.
- at_end (bool): Whether this is the last page of results.
"""
# Use the provided page_size or fallback to the class attribute
page_size = page_size or self.page_size
# Check if there is a cursor
if cursor:
# Get documents with `_id` greater than the cursor
query = {"_id": {'$gt': cursor}}
else:
# Get everything
query = {}
# Sort in ascending order by `_id`
sort_order = 1
# Define the aggregation pipeline
pipeline = [
{"$match": query}, # Filter based on the cursor
{"$sort": {"_id": sort_order}}, # Sort documents by `_id`
{"$limit": page_size + 1}, # Limit results to page_size + 1 to check if there's a next page
# {"$project": {"_id": 1, "title": 1, "content": 1}} # In case you want to return only certain attributes
]
# Execute the aggregation pipeline
results = list(self.cursor_db.aggregate(pipeline))
# logger.debug(results)
# Validate if some data was found
if not results: raise ValueError("No data found")
# Check if there are more documents than the page size
if len(results) > page_size:
# Deleting extra document
results.pop(-1)
# Set the cursor for the next page
next_cursor = results[-1]['_id']
# Set the previous cursor
if cursor:
# in case the cursor have data
prev_cursor = results[0]['_id']
else:
# In case the cursor don't have data (first time)
prev_cursor = None
# Indicate you haven't reached the end of the data
at_end = False
else:
# Indicate that there are not more pages available (last page reached)
next_cursor = None
# Set the cursor for the previous page
prev_cursor = results[0]['_id']
# Indicate you have reached the end of the data
at_end = True
return results, next_cursor, prev_cursor, at_end
def _fetch_previous_page(
self, cursor: ObjectId | None, page_size: int | None = None,
) -> tuple[list, ObjectId | None, ObjectId | None, bool]:
"""Retrieves the previous page of data based on the provided cursor.
Parameters:
-----------
cursor : ObjectId | None
The current cursor indicating the first document of the current page.
page_size : int
The number of documents to retrieve per page.
prev_at_start : bool
Indicates whether the previous page was the first page.
Returns:
--------
tuple:
- results (list): The list of documents retrieved.
- next_cursor (ObjectId | None): The cursor pointing to the start of the next page, None in case is the last page.
- prev_cursor (ObjectId | None): The cursor pointing to the start of the previous page, None in case is the start page.
- at_start (bool): Whether this is the first page of results.
"""
# Use the provided page_size or fallback to the class attribute
page_size = page_size or self.page_size
# Check if there is a cursor
if cursor:
# Get documents with `_id` less than the cursor
query = {'_id': {'$lt': cursor}}
else:
# Get everything
query = {}
# Sort in descending order by `_id`
sort_order = -1
# Define the aggregation pipeline
pipeline = [
{"$match": query}, # Filter based on the cursor
{"$sort": {"_id": sort_order}}, # Sort documents by `_id`
{"$limit": page_size + 1}, # Limit results to page_size + 1 to check if there's a next page
# {"$project": {"_id": 1, "title": 1, "content": 1}} # In case you want to return only certain attributes
]
# Execute the aggregation pipeline
results = list(self.cursor_db.aggregate(pipeline))
# Validate if some data was found
if not results: raise ValueError("No data found")
# Check if there are more documents than the page size
if len(results) > page_size:
# Deleting extra document
results.pop(-1)
# Reverse the results to maintain the correct order
results.reverse()
# Set the cursor for the previous page
prev_cursor = results[0]['_id']
# Set the cursor for the next page
next_cursor = results[-1]['_id']
# Indicate you are not at the start of the data
at_start = False
else:
# Reverse the results to maintain the correct order
results.reverse()
# Indicate that there are not more previous pages available (initial page reached)
prev_cursor = None
# if prev_at_start:
# # in case before was at the starting page
# logger.warning("Caso 1")
# next_cursor = results[0]['_id']
# else:
# # in case before was not at the starting page
# logger.warning("Caso 2")
# next_cursor = results[-1]['_id']
next_cursor = results[-1]['_id']
# Indicate you have reached the start of the data
at_start = True
return results, next_cursor, prev_cursor, at_start
def start_pagination(self):
"""Inicia la navegacion de datos."""
# Change page size in case you want it, only leave it here for reference
page_size = None
# Retrieve the first page of results
results, next_cursor, prev_cursor, at_end = self._fetch_next_page(None, page_size)
at_start = True
logger.info(f"{results = }")
logger.info(f"{next_cursor = }")
logger.info(f"{prev_cursor = }")
logger.info(f"{at_start = }")
logger.info(f"{at_end = }")
# if next_cursor:
if not(at_start and at_end):
while(True):
print(125 * "*")
if at_end:
inn = input("Can only move Backward (b) or Cancel (c): ")
stage = 0
# =====================================================
# You could reset at_end here, but in this example that
# will fail in case the user sends something different
# from Backward (b) or Cancel (c)
# =====================================================
# at_end = False
elif at_start:
inn = input("Can only move Forward (f) or Cancel (c): ")
stage = 1
# =====================================================
# You could reset at_end here, but in this example that
# will fail in case the user sends something different
# from Forward (f) or Cancel (c)
# =====================================================
# at_start = False
else:
inn = input("Can move Forward (f), Backward (b), or Cancel (c): ")
stage = 2
# Execute action acording to the input
if inn == "f" and stage in [1, 2]:
results, next_cursor, prev_cursor, at_end = self._fetch_next_page(next_cursor, page_size)
# For this example, you must reset here the value, otherwise you lose the reference of the cursor
at_start = False
elif inn == "b" and stage in [0, 2]:
# results, next_cursor, prev_cursor, at_start = self._fetch_previous_page(prev_cursor, at_start, page_size)
results, next_cursor, prev_cursor, at_start = self._fetch_previous_page(prev_cursor, page_size)
# For this example, you must reset here the value, otherwise you lose the reference of the cursor
at_end = False
elif inn == "c":
logger.warning("------- Canceling execution -------")
break
else:
print("Not valid action, it can only move in the opposite direction.")
continue
logger.info(f"{results = }")
logger.info(f"{next_cursor = }")
logger.info(f"{prev_cursor = }")
logger.info(f"{at_start = }")
logger.info(f"{at_end = }")
else:
logger.warning("There is not more data to show")
@logger.catch
def main():
"""Main function."""
my_cursor = cursorPattern(page_size=5)
# my_cursor.add_data()
my_cursor.start_pagination()
if __name__:
main()
logger.info("--- Execution end ---")
The page_size was added as an attribute to the class cursorPattern for it to be easier to define the size of every page and added docstrings to the class and its methods.
Hope this will help/guide someone that needs to implement Cursor Pagination.
위 내용은 커서 페이지 매김 예의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
