piccolo-cursor-pagination 0.3.0

Description:

piccolocursorpagination 0.3.0

Cursor pagination for Piccolo ORM
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.
Installation
pip install piccolo_cursor_pagination

Usage
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.
Customization
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")

Directions
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)

Limitations
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 ORM id (PK type integer) column as unique and sequential column for pagination.