Client#

Top-level Subgrounds subpackage

This subpackage implements the top-level API that most developers will be using when

querying subgraphs with Subgrounds. Subgrounds and AsyncSubgrounds contains fully implemented subgraph querying SDKs for accessing subgraph data (flattened or nested) effectively with transformation and pagination bundled in.

SubgroundsBase provides an extended interface to build custom clients on-top allowing

for intricate customization on the buisness logic subgrounds implements.

class subgrounds.client.AsyncSubgrounds(timeout=30, headers=<factory>, global_transforms=<factory>, subgraphs=<factory>, schema_cache=PosixPath('schemas'))#
async load(url, save_schema=False, is_subgraph=True)#

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.

async load_subgraph(url, save_schema=False)#

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) -- Flag indicating whether or not the schema should be cached to disk. Defaults to False.

  • cache_dir -- If save_schema == True, then subgraph schemas will be stored under cache_dir. Defaults to schemas/

Returns:

A generated class representing the subgraph and its entities

Return type:

Subgraph

async load_api(url, save_schema=False)#
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. Defaults to False.

Returns:

A generated class representing the subgraph and its entities

Return type:

Subgraph

async 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

async query_json(fpaths, pagination_strategy=LegacyStrategy)#

See query_json().

Parameters:
Returns:

The reponse data

Return type:

list[dict[str, Any]]

async query_df(fpaths, columns=None, concat=False, pagination_strategy=LegacyStrategy)#

See query_df().

Parameters:
Returns:

A pandas.DataFrame containing the reponse data.

Return type:

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

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

See query().

Parameters:
Returns:

The FieldPath object(s) data

Return type:

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

class subgrounds.client.Subgrounds(timeout=30, headers=<factory>, global_transforms=<factory>, subgraphs=<factory>, schema_cache=PosixPath('schemas'))#
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]

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

A base instance for all Subgrounds (should not be used directly)

classmethod from_pg_key(key, **kwargs)#

Create a Subgrounds* instance using a playgrounds key directly.

This sets the headers field internally to be used with all queries made out.

Parameters:

  • key (str) -- The aforementioned Playgrounds API Key

  • **kwargs (Any) -- Anything else to construct the Subgrounds* instance

Returns:

An instance Subgrounds* with Playgrounds API support baked in

Return type:

Self

make_request(fpaths)#

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

Parameters:

fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more FieldPath objects that should be included in the request

Returns:

Brand new request

Return type:

DataRequest

mk_request(fpaths)#

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

Parameters:

fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more FieldPath objects that should be included in the request

Returns:

Brand new request

Return type:

DataRequest

fetch_schema(url)#

Reads schema from filesystem based on subgraph_slug of the url

cache_schema(url, data)#

Writes schema into filesystem based on subgraph_slug of the url