Hai, saya ingin berkongsi contoh Corak Penomboran Kursor (atau Corak Paging Kursor) kerana semasa saya mencari satu, saya hanya dapat mencari contoh kes yang bergerak ke hadapan tetapi tidak mundur, dan juga cara mengendalikan data pada permulaan dan akhir.
Anda boleh melihat repositori untuk ini di sini tetapi saya akan cuba menerangkan semuanya di sini.
Saya menggunakan Puisi Python sebagai alat pengurusan pakej, jadi untuk contoh ini saya menganggap anda sudah memilikinya. Perkara pertama yang perlu dilakukan ialah memasang kebergantungan dengan pemasangan puisi. Anda juga boleh menggunakan pip untuk memasangnya dengan: pip install pymongo loguru.
Kini kami juga memerlukan pangkalan data Mongo, anda boleh memuat turun Edisi Komuniti MongoDB di sini, dan anda boleh mengkonfigurasinya dengan panduan ini.
Sekarang kita telah memasang kebergantungan dan pangkalan data, kita boleh menambah data padanya. Untuk itu kita boleh menggunakan ini:
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)
Salin selepas log masuk
Dengan itu kami membuat sambungan ke pangkalan data tempatan ke kandungan koleksi. Kemudian kami menambah nilai daripada sample_posts padanya. Sekarang kita mempunyai data untuk dicari, kita boleh mula menanyakannya. Mari kita mula mencari dan membaca data sehingga tamat.
# 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 ---")
Salin selepas log masuk
Kod itu mengembalikan ini:
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 ---
Salin selepas log masuk
Anda boleh melihat bahawa kursor menghala ke halaman seterusnya dan yang sebelumnya ialah Tiada, juga, ia mengenal pasti bahawa ia bukan penghujung data. Untuk mendapatkan nilai ini kita mesti melihat dengan lebih baik pada fungsi fetch_next_page. Di sana kita dapat melihat bahawa kita mentakrifkan page_size, pertanyaan, sort_order, dan kemudian kita mencipta saluran paip ke operasi pengagregatan. Untuk mengenal pasti sama ada terdapat halaman maklumat lain yang kami gunakan operator $limit, kami memberikan nilai page_size + 1 untuk menyemak sama ada terdapat, sebenarnya, halaman lain dengan + 1 itu. Untuk benar-benar menyemaknya kami menggunakan ekspresi len( keputusan) > page_size, sekiranya bilangan data yang dikembalikan lebih besar daripada page_size maka terdapat halaman lain; sebaliknya, ini adalah halaman terakhir.
Untuk kes di mana terdapat halaman seterusnya, kami mesti memadamkan elemen terakhir daripada senarai maklumat yang kami tanya, kerana itu adalah + 1 dalam perancangan, kami perlu menetapkan next_cursor dengan _id daripada nilai terakhir semasa daripada senarai, dan tetapkan prev_cursor (kursor sebelumnya) mengikut kes, jika terdapat kursor yang bermakna terdapat data sebelum ini, dalam kes lain, itu bermakna ini adalah kumpulan data pertama, jadi ada tiada maklumat sebelumnya, oleh itu, kursor harus menjadi _id pertama daripada data yang ditemui atau Tiada.
Sekarang kita tahu cara mencari data dan menambah beberapa pengesahan penting, kita mesti membolehkan cara untuk melintasinya ke hadapan, untuk itu kita akan menggunakan arahan input untuk meminta daripada pengguna yang menjalankan skrip untuk menulis arah untuk bergerak, walaupun, sekarang ia hanya akan ke hadapan (f). Kita boleh mengemas kini fungsi utama kita untuk melakukannya seperti ini:
@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")
Salin selepas log masuk
Dengan ini kami dapat melintasi data sehingga akhir, tetapi apabila ia sampai ke penghujung ia kembali ke permulaan dan kitaran bermula semula, jadi kami mesti menambah beberapa pengesahan untuk mengelakkannya dan juga untuk bergerak ke belakang. Untuk itu kami akan mencipta fungsi fetch_previous_page dan menambah beberapa perubahan pada fungsi utama:
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
Salin selepas log masuk
Sangat serupa dengan fetch_next_page, tetapi pertanyaan (sekiranya syarat dipenuhi) menggunakan operator $lt dan sort_order mestilah -1 untuk membawa data dalam susunan yang diperlukan. Sekarang, apabila mengesahkan jika len(hasil) > page_size, sekiranya keadaannya benar, maka ia mengalih keluar elemen tambahan dan membalikkan susunan data untuk ditunjukkan dengan betul kemudian tetapkan kursor sebelumnya kepada elemen pertama data dan kursor seterusnya hingga yang terakhir. Sebaliknya, data adalah terbalik, kursor sebelumnya ditetapkan kepada Tiada (kerana tiada data sebelumnya), dan tetapkan kursor seterusnya kepada nilai terakhir senarai. Dalam kedua-dua kes, pembolehubah boolean dipanggil at_start ditakrifkan untuk mengenal pasti situasi ini. Sekarang kita mesti menambah interaksi dengan pengguna untuk pergi ke belakang dalam fungsi utama, jadi terdapat 3 situasi untuk dikendalikan sekiranya kita berada di permulaan, penghujung, atau di tengah data: hanya ke hadapan, hanya ke belakang , dan maju ke belakang:
@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")
Salin selepas log masuk
Kami menambah pengesahan pada input pengguna untuk mengenal pasti peringkat di mana kami berada semasa merentasi data, juga ambil perhatian bahawa at_start dan at_end selepas pelaksanaan fetch_next_page dan fetch_previous_page masing-masing yang diperlukan untuk menetapkan semula selepas mencapainya had. Kini anda boleh mencapai penghujung data dan pergi ke belakang sehingga permulaan. Pengesahan selepas mendapat halaman pertama data telah dikemas kini untuk menyemak sama ada bendera at_start dan at_end adalah Benar, yang akan menunjukkan bahawa tiada lagi data untuk ditunjukkan.
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 ---")
Salin selepas log masuk
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.
Atas ialah kandungan terperinci Contoh Penomboran Kursor. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!
![[Web front-end] Permulaan pantas Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
