subgrounds.query module#

Query data structure module

This module contains various data structures in the form of dataclasses that are used to represent GraphQL queries in Subgrounds using an AST-like approach. To the extent possible, these dataclasses are immutable (i.e.: frozen=True) to enforce a functional programming style and reduce side-effects.

A typical Subgrounds request will have the following dataclass hierarchy:

DataRequest
└── Document
    └── Query
        ├── VariableDefinition
        │   └── InputValue
        └── Selection
            ├── Argument
            │   └── InputValue
            └── Selection

class subgrounds.query.InputValue#

Bases: object

class T(*args, **kwargs)#

Bases: Protocol

property graphql: str#

Returns a GraphQL string representation of the input value

Returns:: The GraphQL string representation of the input value
Return type:: str

property is_variable: bool#

Returns True i.f.f. the input value is of type Variable

Returns:: True i.f.f. the input value is of type Variable, otherwise False
Return type:: bool

property is_number: bool#

Returns True i.f.f. the input value is of type Float or Int

Returns:: True i.f.f. the input value is of type Float or Int, otherwise False
Return type:: bool

iter()#

class Null#

Bases: object

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class Int(value: 'int')#

Bases: object

value: int#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class Float(value: 'float')#

Bases: object

value: float#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class String(value: 'str')#

Bases: object

value: str#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class Boolean(value: 'bool')#

Bases: object

value: bool#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class Enum(value: 'str')#

Bases: object

value: str#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class Variable(name: 'str')#

Bases: object

name: str#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class List(value: 'list[InputValue.T]')#

Bases: object

value: list[subgrounds.query.InputValue.T]#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class Object(value: 'dict[str, InputValue.T]')#

Bases: object

value: dict[str, subgrounds.query.InputValue.T]#

property graphql: str#

property is_variable: bool#

property is_number: bool#

iter()#

class subgrounds.query.VariableDefinition(name, type_, default=None)#

Bases: object

Representation of a GraphQL variable definition

name#

Name of the argument

Type:: str

type_#

GraphQL type of the argument

Type:: TypeRef.T

default#

Default value of the variable. Defaults to None.

Type:: InputValue.T, optional

name: str#

type_: T#

default: Optional[T] = None#

property graphql: str#

Returns the GraphQL string representation of the variable definition

Example:

>>> vardef = VariableDefinition(
...   name='foo',
...   type_=TypeRef.NonNull(TypeRef.Named(name="Int", kind="SCALAR")),
...   default=InputValue.Int(100)
... )
>>> print(vardef.graphql)
$foo: Int! = 100

Returns:: The GraphQL string representation of the variable definition
Return type:: str

class subgrounds.query.Argument(name: 'str', value: 'InputValue.T')#

Bases: object

name: str#

value: T#

property graphql: str#

iter()#

iter_vars()#

for_all(predicate)#

for_all_vars(predicate)#

exists(predicate)#

exists_vars(predicate)#

find(predicate)#

find_var(predicate)#

all_defined(variables)#

class subgrounds.query.Selection(fmeta, alias=None, arguments=<factory>, selection=<factory>)#

Bases: object

Represents a GraphQL field selection.

fmeta#

The type definition of the field being selected.

Type:: TypeMeta.FieldMeta

alias#

The alias of the field selection. Defaults to None.

Type:: str, optional

arguments#

The arguments, if any, of the field selection. Defaults to [].

Type:: list[Argument]

selection#

The inner field selections, if any. Defaults to [].

Type:: list[Selection]

fmeta: FieldMeta#

alias: Optional[str] = None#

arguments: list[subgrounds.query.Argument]#

selection: list[subgrounds.query.Selection]#

property key#

property args_graphql: str#

graphql(level=0)#

property data_path: list[str]#

property data_paths: list[list[str]]#

iter()#

Returns an iterator over all Selections of the current selection tree.

iter_args(recurse=True)#

Returns an iterator over all Arguments of the current Selection.

If recurse == True, then the iterator also includes Arguments of inner Selections.

filter(predicate)#

Returns a new Selection object containing all attributes of the current Selection if predicate(self) == True and None otherwise. The function if also applied recursively to inner Selections.

Parameters:: predicate (Callable[[Selection], bool]) -- _description_
Returns:: _description_
Return type:: Optional[Selection]

filter_args(predicate, recurse=True)#

Returns a new Selection object which contains all attributes of the current Selection except for Arguments for which predicate(arg) == True.

If recurse == True, then the function is applied recursively to inner Selections

Parameters:

predicate (Callable[[Argument], bool]) -- _description_
recurse (bool, optional) -- _description_. Defaults to True.

Returns:

_description_

Return type:

Selection

map(map_f, priority='self')#

Returns a new Selection object containing the same selection tree as the current Selection where each Selection object s is map_f(s)

Parameters:: map_f (Callable[[Selection], Selection]) -- Mapping function to apply to each Selection
Returns:: _description_
Return type:: Selection

map_args(map_f, recurse=True)#

Replaces each Argument arg in the current Selection with map_f(arg) and returns a new Selection object containinf the modified arguments.

If recurse == True, then the function is applied recursively to inner Selections.

Parameters:

map_f (Callable[[Argument], Argument | list[Argument]]) -- _description_
recurse (bool, optional) -- _description_. Defaults to True.

Returns:

_description_

Return type:

Selection

filter_map(map_f)#

filter_map_args(map_f, recurse=True)#

for_all(predicate)#

for_all_args(predicate, recurse=True)#

exists(predicate)#

exists_args(predicate, recurse=True)#

find(predicate)#

find_args(predicate, recurse=True)#

find_all(predicate)#

find_all_args(predicate, recurse=True)#

T#: alias of TypeVar('T')

fold(fold_f, parents=[])#

contains_list()#

Returns True i.f.f. the selection self selects a field of type list.

Parameters:: self (Selection) -- The selection to traverse
Returns:: True if selection or nested selections selects a list field. False otherwise.
Return type:: bool

split()#

Returns a list of selections where each of the selections corresponds to a single selection path from the root to a leaf for each leaf selected in self.

Example (simplified, does not show all attributes):

>>> select = Selection('foo', inner=[
...   Selection('bar', inner=[
...     Selection('field0', inner=[]),
...     Selection('field1', inner=[]),
...   ]),
...   Selection('x', inner=[])
... ])
>>> split(select)
[
  Selection('foo', inner=[Selection('bar', inner=[Selection('field0', inner=[])])]),
  Selection('foo', inner=[Selection('bar', inner=[Selection('field1', inner=[])])]),
  Selection('foo', inner=[Selection('x', inner=[])]),
]

Parameters:: self (Selection) -- The selection to split
Returns:: The split selections
Return type:: list[Selection]

extract_data(data)#

add(new_selections)#

Returns a new selection consisting of a copy of self expanded with the selection(s) new_selections. It is assumed that new_selections are inner selections of the root selection self.

Parameters:

self (Selection) -- The Selection object to be expanded
new_selections (Selection | list[Selection]) -- A single or multiple Selection object(s) to be added to self

Returns:

The resulting new selection, i.e.: self: expanded with new_selections

Return type:

Selection

remove(to_remove)#

Returns a new Selection object consisting of a copy of self without the selections in selections_to_remove.

Parameters:

to_remove (Selection | list[Selection]) -- The selection(s) to remove from self

Returns:

The new trimmed down selection, i.e.: self without: selections_to_remove

Return type:

Selection

variable_args(recurse=True)#

Returns all arguments in the current selection which have been given a variable as value.

If recurse == True, then the function is applied recursively to inner selections.

Parameters:: recurse (bool, optional) -- _description_. Defaults to True.
Returns:: _description_
Return type:: list[Argument]

infer_variable_definitions()#

combine(other)#

static merge(selections)#

Returns a list of Selection objects resulting from merging selections to the extent possible.

Parameters:: selections (list[Selection]) -- The selections to be merged
Returns:: _description_
Return type:: list[Selection]

contains(other)#

Returns True i.f.f. the Selection other is a subtree of the Selection self and False otherwise

Parameters:

self (Selection) -- The selection
other (Selection) -- The subselection

Returns:

True i.f.f. other is in self

Return type:

bool

contains_argument(argname, recurse=True)#

Returns True i.f.f. there is an Argument object in self named argname. If recurse is True, then the method also checks the nested selections for an argument named argname.

Parameters:

self (Selection) -- The selection
argname (str) -- The name of the argument
recurse (bool, optional) -- Flag indicating whether or not the method should be run recursively on nested selections. Defaults to True.

Returns:

True i.f.f. there is an argument named argname in self

Return type:

bool

get_argument(argname, recurse=True)#

Returns an Argument object corresponding to the argument in the Selection object select with name argname. If select does not contain such an argument and recurse is True, then the function is called recursively on select's inner selections. If no such argument is found in select or its inner selections, then the function raises an exception.

Parameters:

select (Selection) -- The selection to scan
argname (str) -- The name of the argument to find
recurse (bool, optional) -- Flag indicating whether or not the method should be run recursively on nested selections. Defaults to True.

Raises:

KeyError -- If no argument named argname exists in the selection self.

Returns:

The argument in select with name argname (if any).

Return type:

Argument

get_argument_by_variable(varname, recurse=True)#

Returns an Argument object corresponding to the argument in the Selection object select whose value is a variable named varname. If select does not contain such an argument and recurse is True, then the function is called recursively on select's inner selections. If no such argument is found in select or its inner selections, then the function raises an exception

Parameters:

select (Selection) -- The selection to scan
varname (str) -- The name of the variable to find
recurse (bool, optional) -- Flag indicating whether or not the function should be run recursively. Defaults to True.

Raises:

KeyError -- If no argument with variable value named varname exists in the selection self.

Returns:

The argument in select with variable value named: varname if it exists

Return type:

Argument

substitute_arg(argname, replacement, recurse=True)#

Returns a new Selection object containing the same data as self with the argument named argname replaced with replacement. If recurse is True, then the method is called recursively on self's inner selections and the substitution is also applied to the latter.

Parameters:

self (Selection) -- _description_
argname (str) -- The name of the argument to substitute.
replacement (Argument | list[Argument]) -- The argument(s) replacement
recurse (bool, optional) -- Flag indicating whether or not the method should be run recursively. Defaults to True.

Returns:

_description_

Return type:

Selection

select(other)#

prune_undefined(variables)#

Return a new Selection containing the subtree of the current Selection where all argument InputValues are defined, i.e.: each argument's InputValue is either

1) not of type InputValue.Variable or 2) of type InputValue.Variable and the variable name is contained in variables.

Parameters:: variables (Iterator[str]) -- An iterator over defined variables
Returns:: A new pruned Selection object
Return type:: Selection

class subgrounds.query.Query(name: 'Optional[str]' = None, selection: 'list[Selection]' = <factory>, variables: 'list[VariableDefinition]' = <factory>)#

Bases: object

name: Optional[str] = None#

selection: list[subgrounds.query.Selection]#

variables: list[subgrounds.query.VariableDefinition]#

property graphql: str#

Returns a string containing a GraphQL query matching the current query

Returns:: The string containing the GraphQL query
Return type:: str

iter()#

Returns an iterator over all Selections of the selection tree of the current Query.

iter_args()#

Returns an iterator over all Arguments of the selection tree of the current Query.

iter_vardefs()#

Returns an iterator over all VariableDefinitions of the selection tree of the current Query.

filter(predicate)#

Returns a new Query object containing all selections s that satisfy predicate(s) == True.

Parameters:: predicate (Callable[[Selection], bool]) -- _description_
Returns:: _description_
Return type:: Query

filter_args(predicate)#

Returns a new Query object containing all selections arguments arg that satisfy predicate(arg) == True.

Parameters:: predicate (Callable[[Argument], bool]) -- _description_
Returns:: _description_
Return type:: Query

filter_vardefs(predicate)#

map(map_f, priority='self')#

Applies the function map_f to each Selection in the current Query and returns a new Query object containing the resulting Selections.

Parameters:: map_f (Callable[[Selection], Selection]) -- Mapping function to apply to each Selection
Returns:: _description_
Return type:: Query

map_args(map_f)#

Applies the function map_f to each Argument in the current Query and returns a new Query object containing the resulting Arguments.

Parameters:: map_f (Callable[[Argument], Argument]) -- _description_
Returns:: _description_
Return type:: Selection

map_vardefs(map_f)#

filter_map(map_f)#

filter_map_args(map_f)#

filter_map_vardefs(map_f)#

for_all(predicate)#

for_all_args(predicate)#

for_all_vardefs(predicate)#

exists(predicate)#

exists_args(predicate)#

exists_vardefs(predicate)#

find(predicate)#

find_args(predicate)#

find_vardefs(predicate)#

T#: alias of TypeVar('T')

fold(fold_f)#

infer_variable_definitions()#

add(other)#

Returns a new Query containing all selections in :attr:'self' along with the new selections in other

Parameters:

self (Query) -- The query to which new selection(s) or query are to be added
other (Query | Selection | list[Selection]) -- The new selection(s)
query (or query to be added to the) --

Returns:

A new Query objects containing all selections

Return type:

Query

add_vardefs(vardefs)#

remove(other)#

Returns a new Query object containing all selections in self minus the subquery or selection(s) specified in other.

Note: other does not need to be a "full" selection (i.e.: a selection all the way to leaves of the GraphQL schema).

Example:

>>> og_selection = Selection(TypeMeta.FieldMeta('pair', description="", args=[], type=TypeRef.non_null_list("Pair", kind="OBJECT")), None, [], [
...   Selection(TypeMeta.FieldMeta('token0', description="", args=[], type=TypeRef.Named(name="Token", kind="OBJECT")), None, [], [
...     Selection(TypeMeta.FieldMeta('id', description="", args=[], type=TypeRef.Named(name="String", kind="SCALAR")), None, [], []),
...     Selection(TypeMeta.FieldMeta('name', description="", args=[], type=TypeRef.Named(name="String", kind="SCALAR")), None, [], []),
...     Selection(TypeMeta.FieldMeta('symbol', description="", args=[], type=TypeRef.Named(name="String", kind="SCALAR")), None, [], []),
...   ])
... ])
>>> selection_to_remove = Selection(TypeMeta.FieldMeta('token0', description="", args=[], type=TypeRef.Named(name="Token", kind="OBJECT")), None, [], [])
>>> og_selection.remove(selection_to_remove)
Selection(TypeMeta.FieldMeta('pair', description="", args=[], type=TypeRef.non_null_list("Pair", kind="OBJECT")), None, [], [])

Parameters:

query (Query) -- The query to which a selection has to be removed
other (Query | Selection | list[Selection]) -- The subquery or selection(s) to remove from self

Returns:

A new Query object containing the original query selections: minus other

Return type:

Query

static transform(query, variable_f=<function identity>, selection_f=<function identity>)#

contains_selection(selection)#

Returns True i.f.f. the selection tree selection is present in query.

Parameters:

query (Query) -- A query object
selection (Selection) -- The selection to be found (or not) in query

Returns:

True if the selection is present in query, False: otherwise.

Return type:

bool

contains_argument(argname)#

get_argument(argname)#

static substitute_arg(query, arg_name, replacement)#

static contains(query, other)#

Returns True i.f.f. all selections in other are contained in query. In other words, returns true i.f.f. other is a subset of query.

Note: other does not need to include "full" selections (i.e.: selections all the way to leaves of the GraphQL schema).

Parameters:

query (Query) -- The query that is to be checked
other (Query) -- The query that has to be in query

Returns:

True i.f.f. all selections in other are contained in query, otherwise False

Return type:

bool

static select(query, other)#

Returns a new Query

Parameters:

query (Query) -- [description]
other (Query) -- [description]

Returns:

[description]

Return type:

Query

prune_undefined(variables)#

class subgrounds.query.Fragment(name: 'str', type_: 'TypeRef.T', selection: 'list[Selection]' = <factory>, variables: 'list[VariableDefinition]' = <factory>)#

Bases: object

name: str#

type_: T#

selection: list[subgrounds.query.Selection]#

variables: list[subgrounds.query.VariableDefinition]#

property graphql#

static combine(frag, other)#

static transform(frag, f)#

class subgrounds.query.Document(url: 'str', query: 'Query', fragments: 'list[Fragment]' = <factory>, variables: 'dict[str, Any]' = <factory>)#

Bases: object

url: str#

query: Query#

fragments: list[subgrounds.query.Fragment]#

variables: dict[str, Any]#

property graphql#

static mk_single_query(url, query)#

filter(predicate)#

filter_args(predicate)#

map(map_f)#

Applies the function map_f to each Selection in the current Document and returns a new Document object containing the resulting Selections.

Parameters:: map_f (Callable[[Selection], Selection]) -- Mapping function to apply to each Selection
Returns:: _description_
Return type:: Query

map_args(map_f)#

Applies the function map_f to each Argument in the current Document and returns a new Document object containing the resulting Arguments.

Parameters:: map_f (Callable[[Argument], Argument]) -- _description_
Returns:: _description_
Return type:: Selection

filter_map(map_f)#

static combine(doc, other)#

static transform(doc, query_f=<function identity>, fragment_f=<function identity>)#

prune_undefined(variables)#

Returns a new Document object that contains the subset of the current Document's query containing only the Selections for which all its arguments are defined (i.e.: either constants or variables in variables).

Parameters:: variables (dict[str, Any]) -- _description_
Returns:: _description_
Return type:: Document

class subgrounds.query.DataRequest(documents: 'list[Document]' = <factory>)#

Bases: object

documents: list[subgrounds.query.Document]#

property graphql#

static combine(req, other)#

static transform(req, f)#

static single_query(url, query)#

static single_document(doc)#

static add_documents(self, docs)#

subgrounds.query.selections_of_object(schema, object_)#

Returns generator of Selection objects that selects all non-list fields of GraphQL Object of Interface object_.

Parameters:

schema (SchemaMeta) -- _description_
object (TypeMeta.ObjectMeta | TypeMeta.InterfaceMeta) -- _description_

Yields:

_type_ -- _description_

subgrounds.query.input_value_of_argument(schema, argmeta, value)#

subgrounds.query.arguments_of_field_args(schema, field, args)#