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 ifsave_schema
isTrue
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 ifsave_schema
isTrue
and returns a generated class representing the subgraph with all its entities.- Parameters:
- Returns:
A generated class representing the subgraph and its entities
- Return type:
- 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
isTrue
and returns a generated class representing the GraphQL endpoint with all its entities.
- Performs introspection on the provided GraphQL API
- async execute(req, pagination_strategy=LegacyStrategy)#
Executes a
DataRequest
and returns aDataResponse
.- Parameters:
req (DataRequest) -- The
DataRequest
object to be executedpagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
A
DataResponse
object representing the response- Return type:
- async query_json(fpaths, pagination_strategy=LegacyStrategy)#
See
query_json()
.- Parameters:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more
FieldPath
objects that should be included in the request.pagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
The reponse data
- Return type:
- async query_df(fpaths, columns=None, concat=False, pagination_strategy=LegacyStrategy)#
See
query_df()
.- Parameters:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more FieldPath objects that should be included in the request
merge -- Whether or not to merge resulting dataframes.
pagination_strategy (Optional[Type[PaginationStrategy]]) -- A class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled.
- 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:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- 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
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
The
FieldPath
object(s) data- Return type:
- 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 ifsave_schema
isTrue
and returns a generated class representing the subgraph with all its entities.
- 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
isTrue
and returns a generated class representing the GraphQL endpoint with all its entities.
- Performs introspection on the provided GraphQL API
- execute(req, pagination_strategy=LegacyStrategy)#
Executes a
DataRequest
and returns aDataResponse
.- Parameters:
req (DataRequest) -- The
DataRequest
object to be executed.pagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
A
DataResponse
object representing the response- Return type:
- 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 executedpagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
An iterator over the
DocumentResponse
pages.- Return type:
- ⚠️ 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:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more
FieldPath
objects that should be included in the request.pagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
The reponse data
- Return type:
- query_json_iter(fpaths, pagination_strategy=LegacyStrategy)#
Same as query_json returns an iterator over the response data pages.
- Parameters:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more
FieldPath
objects that should be included in the request.pagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
The reponse data
- Return type:
- 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 returnedfpaths
is a list ofFieldPath
objects that indicate which data must be queried.columns
is an optional argument used to rename the dataframes(s) columns. The length ofcolumns
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 thecolumns
argument).- Parameters:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more FieldPath objects that should be included in the request
columns (Optional[list[str]]) -- The column labels. Defaults to None.
merge -- Whether or not to merge resulting dataframes.
pagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- 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:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- One or more FieldPath objects that should be included in the request
columns -- The column labels. Defaults to None.
merge -- Whether or not to merge resulting dataframes.
pagination_strategy (Optional[Type[PaginationStrategy]]) -- A Class implementing the
PaginationStrategy
Protocol
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- 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 multipleFieldPath
objects are provided).- Parameters:
fpaths (subgrounds.subgraph.fieldpath.FieldPath | list[subgrounds.subgraph.fieldpath.FieldPath]) -- 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
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- Returns:
The
FieldPath
object(s) data- Return 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=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
. IfNone
, then automatic pagination is disabled. Defaults toLegacyStrategy
.
- 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.
- make_request(fpaths)#
Creates a
DataRequest
object by combining one or moreFieldPath
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:
- mk_request(fpaths)#
Creates a
DataRequest
object by combining one or moreFieldPath
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:
- 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