Sync#

Toplevel Subgrounds module

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

class subgrounds.client.sync.Subgrounds(timeout=30, headers=<factory>, global_transforms=<factory>, subgraphs=<factory>, schema_cache=PosixPath('schemas'))#
property _client#

Cached client

load_subgraph(url, save_schema=False, cache_dir=None)#

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:

  • API. (url The url of the) --

  • save_schema (bool) -- Flag indicating whether or not the schema should be cached to disk.

Returns:

A generated class representing the subgraph and its entities

Return type:

Subgraph

load_api(url, save_schema=False, cache_dir=None)#
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) -- Flag indicating whether or not the schema should be saved to disk.

Returns:

A generated class representing the subgraph and its entities

Return type:

Subgraph

execute(req, pagination_strategy=LegacyStrategy)#

Executes a DataRequest and returns a DataResponse.

Parameters:

  • req (DataRequest) -- The DataRequest object to be executed.

  • pagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns:

A DataResponse object representing the response

Return type:

DataResponse

execute_iter(req, pagination_strategy=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]]) -- A Class implementing the PaginationStrategy Protocol. If None, then automatic pagination is disabled. Defaults to LegacyStrategy.

Returns:

An iterator over the DocumentResponse pages.

Return type:

Iterator[DocumentResponse]

⚠️ DOES NOT apply global transforms across multiple documents or their pages.

Since we yield each page as we get it, it's not possible to accurately perform the transforms since we don't collect the pages. This means transforms expecting multiple documents or pages of documents will be inaccurate.

query_json(fpaths, pagination_strategy=LegacyStrategy)#
Equivalent to

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

Parameters:
Returns:

The reponse data

Return type:

list[dict[str, Any]]

query_json_iter(fpaths, pagination_strategy=LegacyStrategy)#

Same as query_json returns an iterator over the response data pages.

Parameters:
Returns:

The reponse data

Return type:

list[dict[str, Any]]

query_df(fpaths, columns=None, concat=False, pagination_strategy=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:
Returns:

A pandas.DataFrame containing the reponse data.

Return type:

pandas.core.frame.DataFrame | list[pandas.core.frame.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=LegacyStrategy)#

Same as query_df except returns an iterator over the response data pages

Parameters:
Returns:

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

Return type:

Iterator[pandas.core.frame.DataFrame | list[pandas.core.frame.DataFrame]]

query(fpaths, unwrap=True, pagination_strategy=LegacyStrategy)#

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

Parameters:
Returns:

The FieldPath object(s) data

Return type:

str | int | float | bool | list | tuple | None

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=LegacyStrategy)#

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

Parameters:

  • fpath -- One or more FieldPath object(s) to query.

  • unwrap (bool) -- 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]]) -- 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[str | int | float | bool | list[Any] | tuple | None]