49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
from typing import Callable, Any, Type
|
|
import strawberry
|
|
from strawberry.annotation import StrawberryAnnotation
|
|
from strawberry.extensions import FieldExtension
|
|
from strawberry.types.arguments import StrawberryArgument
|
|
from strawberry.types.field import StrawberryField
|
|
from graphql_schema.entities.helpers.pagination import get_pagination_window, PaginationWindow
|
|
|
|
class OffsetPagination[Item](FieldExtension):
|
|
|
|
def __init__(self, item_type: Type[Item]):
|
|
super().__init__()
|
|
|
|
self.item_type = item_type
|
|
|
|
def apply(self, field: StrawberryField) -> StrawberryField:
|
|
offset_arg = StrawberryArgument(
|
|
python_name="offset",
|
|
graphql_name="offset",
|
|
type_annotation=StrawberryAnnotation(annotation=int | None),
|
|
default=0,
|
|
)
|
|
|
|
limit_arg = StrawberryArgument(
|
|
python_name="limit",
|
|
graphql_name="limit",
|
|
type_annotation=StrawberryAnnotation(annotation=int),
|
|
default=10,
|
|
)
|
|
|
|
field.arguments.append(offset_arg)
|
|
field.arguments.append(limit_arg)
|
|
|
|
return field
|
|
|
|
async def resolve_async(
|
|
self, next_: Callable[..., Any], source: Any, info: strawberry.Info,
|
|
limit: int, offset: int = 0,
|
|
**kwargs
|
|
) -> PaginationWindow[Item]:
|
|
query = await next_(source, info, **kwargs)
|
|
|
|
return await get_pagination_window(
|
|
query=query,
|
|
item_type=self.item_type,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|