subgrounds.pagination package#

Submodules#

Module contents#

This module contains all code related to automatic pagination.

The pagination module contains the pagination algorithms (both regular and iterative) that make use of PaginationStrategies.

The preprocess and strategties modules implement the currently supported PaginationStrategies: LegacyStrategy and ShallowStrategy.

The utils module contains some generic functions that are useful in the context of pagination.

subgrounds.pagination.generate_pagination_nodes(schema, document)#
class subgrounds.pagination.LegacyStrategy(schema, document)#

Bases: object

schema: SchemaMeta#
arg_generator: LegacyStrategyArgGenerator#
normalized_doc: Document#
step(page_data=None)#
subgrounds.pagination.normalize(schema, document, pagination_nodes)#

Inject various graphql components to "normalize" the query for pagination.

When we paginate a query, we inject custom filtering based on the order by values. We also add GraphQL variables so that PaginationStrategy only need to change those

to perform pagination.

The main process for normalization begins by recursively adjusting Selection nodes

within the Query tree. We only apply the following steps if the node needs to be paginated.

> Note, these steps always check the current selection and will merge new values

and selections onto whats currently there.

  1. Ensure id is on the Selection

  2. Replace first argument value by $firstX

  3. Replace skip argument value by $skipX

  4. With the orderBy (default being id), generate where filtering arguments

  1. These are used to filter out values when paginating

  1. Set where filtering values (deep union / merge)

subgrounds.pagination.paginate_iter(schema, doc, pagination_strategy, headers)#

Executes the request document doc based on the GraphQL schema schema and returns the response as a JSON dictionary.

Parameters:

  • schema (SchemaMeta) -- The GraphQL schema on which the request document is based

  • doc (Document) -- The request document

Returns:

The response data as a JSON dictionary

Return type:

dict[str, Any]

subgrounds.pagination.paginate(schema, doc, pagination_strategy, headers)#

Executes the request document doc based on the GraphQL schema schema and returns the response as a JSON dictionary.

Parameters:

  • schema (SchemaMeta) -- The GraphQL schema on which the request document is based

  • doc (Document) -- The request document

Returns:

The response data as a JSON dictionary

Return type:

dict[str, Any]

exception subgrounds.pagination.PaginationError(message, strategy)#

Bases: RuntimeError

class subgrounds.pagination.PaginationNode(node_idx, filter_field, first_value, skip_value, filter_value, filter_value_type, key_path, inner=<factory>)#

Bases: object

Class representing the pagination config for a single GraphQL list field.

node_idx#

Index of PaginationNode, used to label pagination arguments for this node.

Type:

int

filter_field#

Name of the node's filter field, e.g.: if filter_name is timestamp_gt, then filter_field is timestamp

Type:

str

first_value#

Initial value of the first argument

Type:

int

skip_value#

Initial value of the skip argument

Type:

int

filter_value#

Initial value of the filter argument (i.e.: where: {filter: FILTER_VALUE})

Type:

Any

filter_value_type#

Type of the filter value

Type:

TypeRef.T

key_path#

Location in the list field to which this pagination node refers to in the initial query

Type:

list[str]

inner#

Nested pagination nodes (if any).

Type:

list[PaginationNode]

node_idx: int#
filter_field: str#
first_value: int#
skip_value: int#
filter_value: Any#
filter_value_type: T#
key_path: list[str]#
inner: list[subgrounds.pagination.preprocess.PaginationNode]#
get_vardefs()#

Returns a list of variable definitions corresponding to this pagination node's pagination arguments as well as the variable definitions related to any nested pagination nodes.

Parameters:

self (PaginationNode) -- The current PaginationNode

Returns:

_description_

Return type:

list[VariableDefinition]

class subgrounds.pagination.PaginationStrategy(*args, **kwargs)#

Bases: Protocol

step(page_data=None)#

Returns the new query document and its variables which will be executed to get the next page of data. If this is the first query made as part of the pagination strategy, then page_data will be None.

If pagination should be interupted (e.g.: if enough entities have been queried), then this method should raise a StopPagination exception.

Parameters:

  • page_data (Optional[dict[str, Any]], optional) -- The previous query's response data.

  • (i.e. (If this is the first query) -- the first page of data), then it will be None.

  • None. (Defaults to) --

Returns:

A tuple (doc, vars) where doc is the query document that will be executed to fetch the next page of data and vars are the variables for that document.

Return type:

Tuple[Document, dict[str, Any]]

subgrounds.pagination.prune_doc(document, args)#
class subgrounds.pagination.ShallowStrategy(schema, document)#

Bases: object

schema: SchemaMeta#
arg_generator: ShallowStrategyArgGenerator#
normalized_doc: Document#
step(page_data=None)#