Support the cursors in a storage

Use this guide to add the cursors of RFC 9865 to a storage written as Write a storage explains. The server pages the resources of this storage with cursors once it announces them, as Page the results with cursors explains.

Page with the position

The cursors need a storage that gives stable pages: Paging the results explains what they guarantee. InMemoryStorage gives them. To add the cursors to another storage, set supports_cursors, and page with a cursor when cursor is not None. The position argument then tells which page to return. It is None on the first page.

The storage below extends the one of Write a storage, and pages its resources by identifier. Its server announces neither the filter nor the sort:

Sync

class CursorSQLiteStorage(SQLiteStorage):
    supports_cursors = True

    def search(self, resource_types, search_request, *, position=None):
        types = {resource_type.name: resource_type for resource_type in resource_types}
        [(total,)] = self._select(types, "COUNT(*)")
        if search_request.cursor is None:
            return self._search_index(types, total, search_request)
        return self._search_cursor(types, total, search_request, position)

    def _search_index(self, types, total, search_request):
        count = search_request.count
        page = [-1 if count is None else count, search_request.start_index_0]
        rows = self._select(types, "*", "ORDER BY id LIMIT ? OFFSET ?", page)
        return SearchPage(total, self._to_resources(types, rows))

    def _search_cursor(self, types, total, search_request, position):
        count = search_request.count
        limit = -1 if count is None else count + 1
        if position is None or "after" in position:
            after = "" if position is None else position["after"]
            clause = "AND id > ? ORDER BY id LIMIT ?"
            rows = self._select(types, "*", clause, [after, limit])
            resources = self._to_resources(types, rows[:count])
            more_after, more_before = len(rows) > len(resources), position is not None
        else:
            clause = "AND id < ? ORDER BY id DESC LIMIT ?"
            rows = self._select(types, "*", clause, [position["before"], limit])
            resources = self._to_resources(types, rows[:count][::-1])
            more_after, more_before = True, len(rows) > len(resources)

        result = SearchPage(total, resources)
        if resources and more_after:
            result.next = {"after": resources[-1].id}
        if resources and more_before:
            result.previous = {"before": resources[0].id}
        return result

    def _select(self, types, columns, clause="", parameters=()):
        among = ", ".join("?" * len(types))
        return self.connection.execute(
            f"SELECT {columns} FROM resources WHERE resource_type IN ({among}) {clause}",
            [*types, *parameters],
        ).fetchall()

    def _to_resources(self, types, rows):
        return [self.row_to_resource(types[row["resource_type"]], row) for row in rows]

Async

class AsyncCursorSQLiteStorage(AsyncSQLiteStorage):
    supports_cursors = True

    async def search(self, resource_types, search_request, *, position=None):
        types = {resource_type.name: resource_type for resource_type in resource_types}
        [(total,)] = await self._select(types, "COUNT(*)")
        if search_request.cursor is None:
            return await self._search_index(types, total, search_request)
        return await self._search_cursor(types, total, search_request, position)

    async def _search_index(self, types, total, search_request):
        count = search_request.count
        page = [-1 if count is None else count, search_request.start_index_0]
        rows = await self._select(types, "*", "ORDER BY id LIMIT ? OFFSET ?", page)
        return SearchPage(total, self._to_resources(types, rows))

    async def _search_cursor(self, types, total, search_request, position):
        count = search_request.count
        limit = -1 if count is None else count + 1
        if position is None or "after" in position:
            after = "" if position is None else position["after"]
            clause = "AND id > ? ORDER BY id LIMIT ?"
            rows = await self._select(types, "*", clause, [after, limit])
            resources = self._to_resources(types, rows[:count])
            more_after, more_before = len(rows) > len(resources), position is not None
        else:
            clause = "AND id < ? ORDER BY id DESC LIMIT ?"
            rows = await self._select(types, "*", clause, [position["before"], limit])
            resources = self._to_resources(types, rows[:count][::-1])
            more_after, more_before = True, len(rows) > len(resources)

        result = SearchPage(total, resources)
        if resources and more_after:
            result.next = {"after": resources[-1].id}
        if resources and more_before:
            result.previous = {"before": resources[0].id}
        return result

    async def _select(self, types, columns, clause="", parameters=()):
        connection = await self.connect()
        among = ", ".join("?" * len(types))
        cursor = await connection.execute(
            f"SELECT {columns} FROM resources WHERE resource_type IN ({among}) {clause}",
            [*types, *parameters],
        )
        return list(await cursor.fetchall())

    def _to_resources(self, types, rows):
        return [self.row_to_resource(types[row["resource_type"]], row) for row in rows]

• The position of the next page is {"after": ...}, with the identifier of the last resource of the page. The position of the previous page is {"before": ...}, with the identifier of the first one. A position is a value that JSON can hold.

• The next page selects the rows after this identifier. The previous page selects the rows before it, in the reverse order, and reverses them afterwards.

• The storage reads count + 1 rows to know whether another page follows.

Support the sort

A storage that supports the sort orders the rows by the sort value, then by identifier, so that no two rows are equal. Its positions also hold the sort value of the resource, as sort_value() gives it. A row without sort value comes last when ascending, and first when descending, so the storage writes the condition on the sort value by hand, rather than with a row comparison, which does not handle NULL.

A backend that can only page with offsets does not set supports_cursors, and its server pages by index only. The server refuses to start when its configuration announces the cursors and the storage does not support them.