Piccolo is an great ecosystem that helps you create ASGI apps faster and easier. LimitOffset is the default Piccolo pagination used by Piccolo Admin and Piccolo API. This package contains usage of cursor pagination which is suitable for large data sets and has better performance than LimitOffset
pagination,
but it is strictly optional because it does not work with the Piccolo Admin and Piccolo API.
pip install piccolo_cursor_pagination
Example usage of CursorPagination
:
import typing as t
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from piccolo_api.crud.serializers import create_pydantic_model
from piccolo_cursor_pagination.pagination import CursorPagination
from home.tables import Task
app = FastAPI()
TaskModelOut: t.Any = create_pydantic_model(
table=Task, include_default_columns=True, model_name="TaskModelOut"
)
@app.get("/tasks/", response_model=t.List[TaskModelOut])
async def tasks(
request: Request,
__cursor: t.Optional[str] = None,
__previous: t.Optional[str] = None,
):
try:
previous = request.query_params["__previous"]
paginator = CursorPagination(cursor=__cursor)
rows_result, headers_result = await paginator.get_cursor_rows(
Task, request
)
rows = await rows_result.run()
headers = headers_result
response = JSONResponse(
{"rows": rows[::-1]},
headers={
"next_cursor": headers["cursor"],
},
)
except KeyError:
paginator = CursorPagination(cursor=__cursor)
rows_result, headers_result = await paginator.get_cursor_rows(
Task, request
)
rows = await rows_result.run()
headers = headers_result
response = JSONResponse(
{"rows": rows},
headers={
"next_cursor": headers["cursor"],
},
)
return response
@app.on_event("startup")
async def open_database_connection_pool():
try:
engine = engine_finder()
await engine.start_connection_pool()
except Exception:
print("Unable to connect to the database")
@app.on_event("shutdown")
async def close_database_connection_pool():
try:
engine = engine_finder()
await engine.close_connection_pool()
except Exception:
print("Unable to connect to the database")
The CursorPagination
stores the value of next_cursor
in the response headers.
We can then use the next_cursor
value to get new set of results by passing
next_cursor
to __cursor
query parameter.
Full Piccolo ASGI app is in example folder or you can check another example with Vue frontend.
The CursorPagination
class has a default value of page_size
and order_by
,
but we can overide this value in constructor to adjust the way the results are displayed.
Example of displaying results in ascending order and page size is 10:
paginator = CursorPagination(cursor=__cursor, page_size=10, order_by="id")
The CursorPagination
has the ability to move forward and backward.
To go backward we have to pass __previous=yes
in the query parameters.
Example usage of direction:
GET http://localhost:8000/tasks/?__cursor=NA== (forward)
GET http://localhost:8000/tasks/?__cursor=NA==&__previous=yes (backward)
You need to be aware that cursor pagination has several trade offs. The cursor must be based on a unique and sequential column in the table and the client can't go to a specific page because there is no concept of the total number of pages or results.
WARNING:
CursorPagination
use Piccolo ORMid
(PK type integer) column as unique and sequential column for pagination.