subgrounds.subgrounds module

Toplevel Subgrounds module

This module implements the toplevel API that most developers will be using when querying The Graph with Subgrounds.

class subgrounds.subgrounds.Subgrounds(global_transforms: list[subgrounds.transform.RequestTransform] = <factory>, subgraphs: dict[str, subgrounds.subgraph.subgraph.Subgraph] = <factory>)

Bases: object

global_transforms: list[subgrounds.transform.RequestTransform]

subgraphs: dict[str, subgrounds.subgraph.subgraph.Subgraph]

load_subgraph(url, save_schema=False, cache_dir='schemas/')

Performs introspection on the provided GraphQL API url to get the schema, stores the schema if save_schema is True and returns a generated class representing the subgraph with all its entities.

Parameters

url (str) -- The url of the API
save_schema (bool, optional) -- Flag indicating whether or not the schema should be saved to disk. Defaults to False.

Returns

A generated class representing the subgraph and its entities

Return type

Subgraph

load_api(url, save_schema=False, cache_dir='schemas/')

Performs introspection on the provided GraphQL API url to get the schema, stores the schema if save_schema is True and returns a generated class representing the GraphQL endpoint with all its entities.

Parameters

url (str) -- The url of the API
save_schema (bool, optional) -- Flag indicating whether or not the schema should be saved to disk. Defaults to False.

Returns

A generated class representing the subgraph and its entities

Return type

Subgraph

mk_request(fpaths)

Creates a DataRequest object by combining one or more FieldPath objects.

Parameters: fpaths (FieldPath | list[FieldPath]) -- One or more FieldPath objects that should be included in the request
Returns: A new DataRequest object
Return type: DataRequest

execute(req, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Executes a DataRequest object, sending the underlying query(ies) to the server and returning a data blob (list of Python dictionaries, one per actual query).

Parameters

req (DataRequest) -- The DataRequest object to be executed
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

The reponse data

Return type

list[dict]

execute_iter(req, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Same as execute, except that an iterator is returned which will iterate the data pages.

Parameters

req (DataRequest) -- The DataRequest object to be executed
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

An iterator over the reponse data pages

Return type

Iterator[dict]

query_json(fpaths, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Equivalent to Subgrounds.execute(Subgrounds.mk_request(fpaths), pagination_strategy).

Parameters

fpaths (FieldPath | list[FieldPath]) -- One or more FieldPath objects that should be included in the request.
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

The reponse data

Return type

list[dict[str, Any]]

query_json_iter(fpaths, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Same as query_json except an iterator over the response data pages is returned.

Parameters

fpaths (FieldPath | list[FieldPath]) -- One or more FieldPath objects that should be included in the request.
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

The reponse data

Return type

list[dict[str, Any]]

query_df(fpaths, columns=None, concat=False, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Same as Subgrounds.query() but formats the response data into a Pandas DataFrame. If the response data cannot be flattened to a single query (e.g.: when querying multiple list fields that return different entities), then multiple dataframes are returned

fpaths is a list of FieldPath objects that indicate which data must be queried.

columns is an optional argument used to rename the dataframes(s) columns. The length of columns must be the same as the number of columns of all returned dataframes.

concat indicates whether or not the resulting dataframes should be concatenated together. The dataframes must have the same number of columns, as well as the same column names and types (the names can be set using the columns argument).

Parameters

fpaths (FieldPath | list[FieldPath]) -- One or more FieldPath objects that should be included in the request.
columns (Optional[list[str]], optional) -- The column labels. Defaults to None.
merge (bool, optional) -- Whether or not to merge resulting dataframes.
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

A DataFrame containing the reponse data

Return type

pd.DataFrame | list[pd.DataFrame]

Example:

>>> from subgrounds import Subgrounds
>>> sg = Subgrounds()
>>> univ3 = sg.load_subgraph('https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3')

# Define price SyntheticField
>>> univ3.Swap.price = abs(univ3.Swap.amount0) / abs(univ3.Swap.amount1)

# Query last 10 swaps from the ETH/USDC pool
>>> eth_usdc = univ3.Query.swaps(
...     orderBy=univ3.Swap.timestamp,
...     orderDirection='desc',
...     first=10,
...     where=[
...         univ3.Swap.pool == '0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8'
...     ]
... )
>>> sg.query_df([
...     eth_usdc.timestamp,
...     eth_usdc.price
... ])
  swaps_timestamp  swaps_price
0       1643213811  2618.886394
1       1643213792  2618.814281
2       1643213792  2617.500494
3       1643213763  2615.458495
4       1643213763  2615.876574
5       1643213739  2615.352390
6       1643213678  2615.205713
7       1643213370  2614.115746
8       1643213210  2613.077301
9       1643213196  2610.686563

query_df_iter(fpaths, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Same as query_df except an iterator over the response data pages is returned :param fpaths: One or more FieldPath objects that

should be included in the request

Parameters

columns (Optional[list[str]], optional) -- The column labels. Defaults to None.
merge (bool, optional) -- Whether or not to merge resulting dataframes.
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

An iterator over the response data pages, each as a DataFrame

Return type

Iterator[pd.DataFrame]

query(fpaths, unwrap=True, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Executes one or multiple FieldPath objects immediately and return the data (as a tuple if multiple FieldPath objects are provided).

Parameters

fpaths (FieldPath | list[FieldPath]) -- One or more FieldPath object(s) to query.
unwrap (bool, optional) -- Flag indicating whether or not, in the case where the returned data is a list of one element, the element itself should be returned instead of the list. Defaults to True.
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

The FieldPath object(s) data

Return type

[type]

Example:

>>> from subgrounds import Subgrounds
>>> sg = Subgrounds()
>>> univ3 = sg.load_subgraph('https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3')

# Define price SyntheticField
>>> univ3.Swap.price = abs(univ3.Swap.amount0) / abs(univ3.Swap.amount1)

# Construct FieldPath to get price of last swap on ETH/USDC pool
>>> eth_usdc_last = univ3.Query.swaps(
...     orderBy=univ3.Swap.timestamp,
...     orderDirection='desc',
...     first=1,
...     where=[
...         univ3.Swap.pool == '0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8'
...     ]
... ).price

# Query last price FieldPath
>>> sg.query(eth_usdc_last)
2628.975030015892

query_iter(fpaths, unwrap=True, pagination_strategy=<class 'subgrounds.pagination.strategies.LegacyStrategy'>)

Same as query except an iterator over the resonse data pages is returned.

Parameters

fpath (FieldPath | list[FieldPath]) -- One or more FieldPath object(s) to query.
unwrap (bool, optional) -- Flag indicating whether or not, in the case where the returned data is a list of one element, the element itself should be returned instead of the list. Defaults to True.
pagination_strategy (Optional[Type[PaginationStrategy]], optional) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns

An iterator over the FieldPath object(s)' data pages

Return type

Iterator[type]