Pagination

edgy offers built-in support for both counter-based and cursor-based high-performance pagination. You can also set extra attributes to make all items behave like a double-linked list.

High-performance means smart caching is used, so if you reuse the paginator, you might even skip database access. This works because it reuses the order of the QuerySet, which may already have the entire query cached.

However, edgy is not as flexible as Django's Paginator. It only accepts QuerySets.

Counter-based

This is the classic way of pagination. You provide a page number, and it returns a specific set of items based on the order of the QuerySet.

import edgy
from edgy.contrib.pagination import Paginator, Page
from edgy import Registry

models = Registry(database="sqlite:///db.sqlite")


class BlogEntry(edgy.Model):
    title: str = edgy.fields.CharField(max_length=100)
    content: str = edgy.fields.CharField(max_length=100)
    created = edgy.fields.DateTimeField(auto_now_add=True)

    class Meta:
        registry = models


async def get_last_blogpost_page() -> tuple[Page, int]:
    # order by is required for paginators
    paginator = Paginator(BlogEntry.query.order_by("-created", "-id"), page_size=30)
    return await paginator.get_page(-1), await paginator.get_amount_pages()


async def get_blogpost_pages() -> tuple[Page, int]:
    # order by is required for paginators
    paginator = Paginator(BlogEntry.query.order_by("-created", "-id"), page_size=30)
    return [page async for page in paginator.paginate()]


async def search_blogpost(title: str, page: int) -> tuple[Page, int]:
    # order by is required for paginators
    paginator = Paginator(
        BlogEntry.query.filter(title__icontains=title).order_by("-created", "-id"), page_size=30
    )
    return await paginator.get_page(page), await paginator.get_amount_pages()

You can also use attributes to get the previous or next item. For better performance, we use the CursorPaginator:

from __future__ import annotations
import edgy
from edgy.contrib.pagination import CursorPaginator
from edgy import Registry

models = Registry(database="sqlite:///db.sqlite")


class BlogEntry(edgy.Model):
    title: str = edgy.fields.CharField(max_length=100)
    content: str = edgy.fields.CharField(max_length=100)
    created = edgy.fields.DateTimeField(auto_now_add=True)

    class Meta:
        registry = models


async def get_blogpost(id: int) -> BlogEntry | None:
    query = BlogEntry.query.order_by("-created", "-id")
    # order by is required for paginators
    paginator = CursorPaginator(
        query,
        page_size=1,
        next_item_attr="next_blogpost",
        previous_item_attr="last_blogpost",
    )
    try:
        created = (await query.get(id=id)).created
        # cursor must match order_by order
        page = await paginator.get_page(cursor=(created, id))
        if page.content:
            return page.content[0]
    except edgy.ObjectNotFound:
        ...
    # get first blogpost as fallback
    fallback_page = await paginator.get_page()
    if fallback_page.content:
        return fallback_page.content[0]
    return None

This example would be in the slow variant:

from __future__ import annotations
import edgy
from edgy.contrib.pagination import Paginator, CursorPaginator
from edgy import Registry

models = Registry(database="sqlite:///db.sqlite")


class BlogEntry(edgy.Model):
    title: str = edgy.fields.CharField(max_length=100)
    content: str = edgy.fields.CharField(max_length=100)
    created = edgy.fields.DateTimeField(auto_now_add=True)

    class Meta:
        registry = models


async def get_blogpost(id: int) -> BlogEntry | None:
    query = BlogEntry.query.order_by("-created", "-id")
    # order by is required for paginators
    paginator = Paginator(
        query,
        page_size=30,
        next_item_attr="next_blogpost",
        previous_item_attr="last_blogpost",
    )
    # this is less performant than the cursor variant
    async for page in paginator.paginate():
        for blogpost in page:
            if blogpost.id == id:
                return blogpost
    # get first blogpost as fallback
    fallback_page = await paginator.get_page()
    if fallback_page.content:
        return fallback_page.content[0]
    return None

Note

If you use a StrictModel make sure you have placeholders in place.

Cursor-based

This pagination works like the counter-based one but supports only one column: It is used as a cursor. This is more efficient and allows querying for new contents, in case of sequential cursors.

from __future__ import annotations

import datetime
import edgy
from edgy.contrib.pagination import CursorPaginator
from edgy import Registry

models = Registry(database="sqlite:///db.sqlite")


class BlogEntry(edgy.Model):
    title: str = edgy.fields.CharField(max_length=100)
    content: str = edgy.fields.CharField(max_length=100)
    created = edgy.fields.DateTimeField(auto_now_add=True)

    class Meta:
        registry = models


async def get_blogpost(cursor: datetime.datetime) -> BlogEntry | None:
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.order_by("-created"),
        page_size=1,
        next_item_attr="next_blogpost",
        previous_item_attr="last_blogpost",
    )
    page = await paginator.get_page(cursor)
    if page.content:
        return page.content[0]
    return None


async def get_next_blogpost_page(cursor: datetime.datetime):
    # order by is required for paginators
    paginator = CursorPaginator(BlogEntry.query.order_by("-created"), page_size=30)
    return await paginator.get_page(cursor), await paginator.get_amount_pages()


async def get_last_blogpost_page(cursor: datetime.datetime):
    # order by is required for paginators
    paginator = CursorPaginator(BlogEntry.query.order_by("-created"), page_size=30)
    return await paginator.get_page(cursor, backward=True), await paginator.get_amount_pages()


async def get_blogpost_pages(after: datetime.datetime | None = None):
    # order by is required for paginators
    paginator = CursorPaginator(BlogEntry.query.order_by("-created"), page_size=30)
    return [page async for page in paginator.paginate(start_cursor=after)]


async def search_blogpost(title: str, cursor: datetime.datetime | None = None):
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.filter(title__icontains=title).order_by("-created"), page_size=30
    )
    return await paginator.get_page(cursor), await paginator.get_amount_pages()

Because you can have vectors as cursors, you can also use this paginator to calculate efficiently the partners for a single item like shown above:

from __future__ import annotations
import edgy
from edgy.contrib.pagination import CursorPaginator
from edgy import Registry

models = Registry(database="sqlite:///db.sqlite")


class BlogEntry(edgy.Model):
    title: str = edgy.fields.CharField(max_length=100)
    content: str = edgy.fields.CharField(max_length=100)
    created = edgy.fields.DateTimeField(auto_now_add=True)

    class Meta:
        registry = models


async def get_blogpost(id: int) -> BlogEntry | None:
    query = BlogEntry.query.order_by("-created", "-id")
    # order by is required for paginators
    paginator = CursorPaginator(
        query,
        page_size=1,
        next_item_attr="next_blogpost",
        previous_item_attr="last_blogpost",
    )
    try:
        created = (await query.get(id=id)).created
        # cursor must match order_by order
        page = await paginator.get_page(cursor=(created, id))
        if page.content:
            return page.content[0]
    except edgy.ObjectNotFound:
        ...
    # get first blogpost as fallback
    fallback_page = await paginator.get_page()
    if fallback_page.content:
        return fallback_page.content[0]
    return None

Integration

How would an application look like, using this feature?

Here an example for esmerald with cursors and attributes:

from typing import Optional
from esmerald import Esmerald, Gateway, get, post
from pydantic import BaseModel

import edgy
from edgy.testing.factory import ModelFactory
from edgy.contrib.pagination import CursorPaginator
from edgy.core.marshalls import Marshall
from edgy.core.marshalls.config import ConfigMarshall

models = edgy.Registry(database="DATABASE_URL")


class BlogEntryBase(edgy.Model):
    # explicit required, otherwise the id is not found because the model is abstract
    id: int = edgy.fields.BigIntegerField(autoincrement=True, primary_key=True)
    title: str = edgy.fields.CharField(max_length=100)
    content: str = edgy.fields.TextField()

    class Meta:
        abstract = True


class BlogEntry(BlogEntryBase):
    # get rid of nested next and last when serializing otherwise we have loops
    next: Optional["BlogEntryBase"] = None
    last: Optional["BlogEntryBase"] = None

    class Meta:
        registry = models


class BlogEntryMarshall(Marshall):
    marshall_config = ConfigMarshall(model=BlogEntry, exclude=["next", "last"])


class BlogEntryFactory(ModelFactory):
    class Meta:
        model = BlogEntry


class BlogPage(BaseModel):
    content: list[BlogEntry]
    is_first: bool
    is_last: bool
    current_cursor: Optional[int]
    next_cursor: Optional[int]
    pages: int


@get("/blog/item/{id}")
async def get_blogpost(id: int) -> Optional["BlogEntry"]:
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.order_by("-id"),
        page_size=1,
        next_item_attr="next",
        previous_item_attr="last",
    )
    page = await paginator.get_page(id)
    if page.content:
        return page.content[0]
    return None


@get("/")
async def index() -> BlogPage:
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.order_by("-id"),
        page_size=30,
        next_item_attr="next",
        previous_item_attr="last",
    )
    p, amount = await paginator.get_page(), await paginator.get_amount_pages()
    # model_dump would also serialize the BlogEntries, so use __dict__ which should be also faster
    return BlogPage(**p.__dict__, pages=amount)


@get("/blog/nextpage/{advance_cursor}")
async def get_next_blogpost_page(advance_cursor: int) -> BlogPage:
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.order_by("-id"),
        page_size=30,
        next_item_attr="next",
        previous_item_attr="last",
    )
    p, amount = await paginator.get_page(advance_cursor), await paginator.get_amount_pages()
    # model_dump would also serialize the BlogEntries, so use __dict__ which should be also faster
    return BlogPage(**p.__dict__, pages=amount)


@get("/blog/lastpage/{reverse_cursor}")
async def get_last_blogpost_page(reverse_cursor: int) -> BlogPage:
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.order_by("-id"),
        page_size=30,
        next_item_attr="next",
        previous_item_attr="last",
    )
    p, amount = (
        await paginator.get_page(reverse_cursor, backward=True),
        await paginator.get_amount_pages(),
    )
    # model_dump would also serialize the BlogEntries, so use __dict__ which should be also faster
    return BlogPage(**p.__dict__, pages=amount)


@post("/search")
async def search_blogpost(string: str, cursor: Optional[int] = None) -> BlogPage:
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.or_({"title__icontains": string}, {"content__icontains": string}).order_by(
            "-id"
        ),
        page_size=30,
        next_item_attr="next",
        previous_item_attr="last",
    )
    p, amount = await paginator.get_page(cursor), await paginator.get_amount_pages()
    # model_dump would also serialize the BlogEntries, so use __dict__ which should be also faster
    return BlogPage(**p.__dict__, pages=amount)


@post("/create")
async def create_blog_entry(data: BlogEntryMarshall) -> BlogEntryMarshall:
    await data.save()
    return data


def get_application():
    app = Esmerald(
        routes=[
            Gateway(handler=create_blog_entry),
            index,
            get_blogpost,
            search_blogpost,
            get_last_blogpost_page,
            get_next_blogpost_page,
        ],
        on_startup=[models.__aenter__],
        on_shutdown=[models.__aexit__],
    )
    return app

Special features

Single-page mode (linked lists)

If you set the page_size to 0, all items are displayed on one page. This transforms the QuerySet into a linked list, where each item knows its neighbors.

The CursorPaginator works a bit different: it shows only one page, but you can still pass cursors to limit the range.

from __future__ import annotations
import datetime
import edgy
from edgy.contrib.pagination import Paginator, CursorPaginator
from edgy import Registry

models = Registry(database="sqlite:///db.sqlite")


class BlogEntry(edgy.Model):
    title: str = edgy.fields.CharField(max_length=100)
    content: str = edgy.fields.CharField(max_length=100)
    created = edgy.fields.DateTimeField(auto_now_add=True)

    class Meta:
        registry = models


async def get_blogposts_with_partners() -> list[BlogEntry]:
    # order by is required for paginators
    paginator = Paginator(
        BlogEntry.query.order_by("-created", "-id"),
        page_size=0,
        next_item_attr="next_blogpost",
        previous_item_attr="last_blogpost",
    )
    return (await paginator.get_page()).content


async def get_blogposts_with_partners_after(after: datetime.datetime) -> list[BlogEntry]:
    # order by is required for paginators
    paginator = CursorPaginator(
        BlogEntry.query.order_by("-created"),
        page_size=0,
        next_item_attr="next_blogpost",
        previous_item_attr="last_blogpost",
    )
    return (await paginator.get_page(after)).content

Reversing

Every paginator has a get_reverse_paginator() method which returns a cached paginator which contains a reversed QuerySet of the current paginator (the order is reversed).

Cache management

Sometimes you need to clear the cache to get fresh results. For this the paginator provides the clear_caches() method.