summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/pagination.py
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/litestar/pagination.py
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/pagination.py')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/pagination.py342
1 files changed, 0 insertions, 342 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/pagination.py b/venv/lib/python3.11/site-packages/litestar/pagination.py
deleted file mode 100644
index 294a13a..0000000
--- a/venv/lib/python3.11/site-packages/litestar/pagination.py
+++ /dev/null
@@ -1,342 +0,0 @@
-# ruff: noqa: UP006,UP007
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass
-from typing import Generic, List, Optional, TypeVar
-from uuid import UUID
-
-__all__ = (
- "AbstractAsyncClassicPaginator",
- "AbstractAsyncCursorPaginator",
- "AbstractAsyncOffsetPaginator",
- "AbstractSyncClassicPaginator",
- "AbstractSyncCursorPaginator",
- "AbstractSyncOffsetPaginator",
- "ClassicPagination",
- "CursorPagination",
- "OffsetPagination",
-)
-
-
-T = TypeVar("T")
-C = TypeVar("C", int, str, UUID)
-
-
-@dataclass
-class ClassicPagination(Generic[T]):
- """Container for data returned using limit/offset pagination."""
-
- __slots__ = ("items", "page_size", "current_page", "total_pages")
-
- items: List[T]
- """List of data being sent as part of the response."""
- page_size: int
- """Number of items per page."""
- current_page: int
- """Current page number."""
- total_pages: int
- """Total number of pages."""
-
-
-@dataclass
-class OffsetPagination(Generic[T]):
- """Container for data returned using limit/offset pagination."""
-
- __slots__ = ("items", "limit", "offset", "total")
-
- items: List[T]
- """List of data being sent as part of the response."""
- limit: int
- """Maximal number of items to send."""
- offset: int
- """Offset from the beginning of the query.
-
- Identical to an index.
- """
- total: int
- """Total number of items."""
-
-
-@dataclass
-class CursorPagination(Generic[C, T]):
- """Container for data returned using cursor pagination."""
-
- __slots__ = ("items", "results_per_page", "cursor", "next_cursor")
-
- items: List[T]
- """List of data being sent as part of the response."""
- results_per_page: int
- """Maximal number of items to send."""
- cursor: Optional[C]
- """Unique ID, designating the last identifier in the given data set.
-
- This value can be used to request the "next" batch of records.
- """
-
-
-class AbstractSyncClassicPaginator(ABC, Generic[T]):
- """Base paginator class for sync classic pagination.
-
- Implement this class to return paginated result sets using the classic pagination scheme.
- """
-
- @abstractmethod
- def get_total(self, page_size: int) -> int:
- """Return the total number of records.
-
- Args:
- page_size: Maximal number of records to return.
-
- Returns:
- An integer.
- """
- raise NotImplementedError
-
- @abstractmethod
- def get_items(self, page_size: int, current_page: int) -> list[T]:
- """Return a list of items of the given size 'page_size' correlating with 'current_page'.
-
- Args:
- page_size: Maximal number of records to return.
- current_page: The current page of results to return.
-
- Returns:
- A list of items.
- """
- raise NotImplementedError
-
- def __call__(self, page_size: int, current_page: int) -> ClassicPagination[T]:
- """Return a paginated result set.
-
- Args:
- page_size: Maximal number of records to return.
- current_page: The current page of results to return.
-
- Returns:
- A paginated result set.
- """
- total_pages = self.get_total(page_size=page_size)
-
- items = self.get_items(page_size=page_size, current_page=current_page)
-
- return ClassicPagination[T](
- items=items, total_pages=total_pages, page_size=page_size, current_page=current_page
- )
-
-
-class AbstractAsyncClassicPaginator(ABC, Generic[T]):
- """Base paginator class for async classic pagination.
-
- Implement this class to return paginated result sets using the classic pagination scheme.
- """
-
- @abstractmethod
- async def get_total(self, page_size: int) -> int:
- """Return the total number of records.
-
- Args:
- page_size: Maximal number of records to return.
-
- Returns:
- An integer.
- """
- raise NotImplementedError
-
- @abstractmethod
- async def get_items(self, page_size: int, current_page: int) -> list[T]:
- """Return a list of items of the given size 'page_size' correlating with 'current_page'.
-
- Args:
- page_size: Maximal number of records to return.
- current_page: The current page of results to return.
-
- Returns:
- A list of items.
- """
- raise NotImplementedError
-
- async def __call__(self, page_size: int, current_page: int) -> ClassicPagination[T]:
- """Return a paginated result set.
-
- Args:
- page_size: Maximal number of records to return.
- current_page: The current page of results to return.
-
- Returns:
- A paginated result set.
- """
- total_pages = await self.get_total(page_size=page_size)
-
- items = await self.get_items(page_size=page_size, current_page=current_page)
-
- return ClassicPagination[T](
- items=items, total_pages=total_pages, page_size=page_size, current_page=current_page
- )
-
-
-class AbstractSyncOffsetPaginator(ABC, Generic[T]):
- """Base paginator class for limit / offset pagination.
-
- Implement this class to return paginated result sets using the limit / offset pagination scheme.
- """
-
- @abstractmethod
- def get_total(self) -> int:
- """Return the total number of records.
-
- Returns:
- An integer.
- """
- raise NotImplementedError
-
- @abstractmethod
- def get_items(self, limit: int, offset: int) -> list[T]:
- """Return a list of items of the given size 'limit' starting from position 'offset'.
-
- Args:
- limit: Maximal number of records to return.
- offset: Starting position within the result set (assume index 0 as starting position).
-
- Returns:
- A list of items.
- """
- raise NotImplementedError
-
- def __call__(self, limit: int, offset: int) -> OffsetPagination[T]:
- """Return a paginated result set.
-
- Args:
- limit: Maximal number of records to return.
- offset: Starting position within the result set (assume index 0 as starting position).
-
- Returns:
- A paginated result set.
- """
- total = self.get_total()
-
- items = self.get_items(limit=limit, offset=offset)
-
- return OffsetPagination[T](items=items, total=total, offset=offset, limit=limit)
-
-
-class AbstractAsyncOffsetPaginator(ABC, Generic[T]):
- """Base paginator class for limit / offset pagination.
-
- Implement this class to return paginated result sets using the limit / offset pagination scheme.
- """
-
- @abstractmethod
- async def get_total(self) -> int:
- """Return the total number of records.
-
- Returns:
- An integer.
- """
- raise NotImplementedError
-
- @abstractmethod
- async def get_items(self, limit: int, offset: int) -> list[T]:
- """Return a list of items of the given size 'limit' starting from position 'offset'.
-
- Args:
- limit: Maximal number of records to return.
- offset: Starting position within the result set (assume index 0 as starting position).
-
- Returns:
- A list of items.
- """
- raise NotImplementedError
-
- async def __call__(self, limit: int, offset: int) -> OffsetPagination[T]:
- """Return a paginated result set.
-
- Args:
- limit: Maximal number of records to return.
- offset: Starting position within the result set (assume index 0 as starting position).
-
- Returns:
- A paginated result set.
- """
- total = await self.get_total()
- items = await self.get_items(limit=limit, offset=offset)
-
- return OffsetPagination[T](items=items, total=total, offset=offset, limit=limit)
-
-
-class AbstractSyncCursorPaginator(ABC, Generic[C, T]):
- """Base paginator class for sync cursor pagination.
-
- Implement this class to return paginated result sets using the cursor pagination scheme.
- """
-
- @abstractmethod
- def get_items(self, cursor: C | None, results_per_page: int) -> tuple[list[T], C | None]:
- """Return a list of items of the size 'results_per_page' following the given cursor, if any,
-
- Args:
- cursor: A unique identifier that acts as the 'cursor' after which results should be given.
- results_per_page: A maximal number of results to return.
-
- Returns:
- A tuple containing the result set and a new cursor that marks the last record retrieved.
- The new cursor can be used to ask for the 'next_cursor' batch of results.
- """
- raise NotImplementedError
-
- def __call__(self, cursor: C | None, results_per_page: int) -> CursorPagination[C, T]:
- """Return a paginated result set given an optional cursor (unique ID) and a maximal number of results to return.
-
- Args:
- cursor: A unique identifier that acts as the 'cursor' after which results should be given.
- results_per_page: A maximal number of results to return.
-
- Returns:
- A paginated result set.
- """
- items, new_cursor = self.get_items(cursor=cursor, results_per_page=results_per_page)
-
- return CursorPagination[C, T](
- items=items,
- results_per_page=results_per_page,
- cursor=new_cursor,
- )
-
-
-class AbstractAsyncCursorPaginator(ABC, Generic[C, T]):
- """Base paginator class for async cursor pagination.
-
- Implement this class to return paginated result sets using the cursor pagination scheme.
- """
-
- @abstractmethod
- async def get_items(self, cursor: C | None, results_per_page: int) -> tuple[list[T], C | None]:
- """Return a list of items of the size 'results_per_page' following the given cursor, if any,
-
- Args:
- cursor: A unique identifier that acts as the 'cursor' after which results should be given.
- results_per_page: A maximal number of results to return.
-
- Returns:
- A tuple containing the result set and a new cursor that marks the last record retrieved.
- The new cursor can be used to ask for the 'next_cursor' batch of results.
- """
- raise NotImplementedError
-
- async def __call__(self, cursor: C | None, results_per_page: int) -> CursorPagination[C, T]:
- """Return a paginated result set given an optional cursor (unique ID) and a maximal number of results to return.
-
- Args:
- cursor: A unique identifier that acts as the 'cursor' after which results should be given.
- results_per_page: A maximal number of results to return.
-
- Returns:
- A paginated result set.
- """
- items, new_cursor = await self.get_items(cursor=cursor, results_per_page=results_per_page)
-
- return CursorPagination[C, T](
- items=items,
- results_per_page=results_per_page,
- cursor=new_cursor,
- )