fetchers#

FetcherClient#

class pragma_sdk.common.fetchers.fetcher_client.FetcherClient[source]#

Bases: object

This client extends the pragma client with functionality for fetching from our third party sources. It can be used to synchronously or asynchronously fetch assets.

The client works by setting up fetchers that are provided the assets to fetch and the publisher name.

Example usage:

pairs = [
    Pair.from_tickers("BTC", "USD"),
    Pair.from_tickers("ETH", "USD"),
]

bitstamp_fetcher = BitstampFetcher(pairs, "publisher_test")
gateio_fetcher = GateIOFetcher(pairs, "publisher_test")

fetchers = [
    bitstamp_fetcher,
    gateio_fetcher,
]

fc = FetcherClient()
fc.add_fetchers(fetchers)

await fc.fetch()
fc.fetch_sync()

You can also set a custom timeout duration as followed:

await fc.fetch(timeout_duration=20)  # Denominated in seconds (default=10)

add_fetcher(fetcher: FetcherInterfaceT) → None[source]#: Add a single fetcher to the supported fetchers list.

add_fetchers(fetchers: List[FetcherInterfaceT]) → None[source]#: Add fetchers to the supported fetchers list.

async fetch(filter_exceptions: bool = True, return_exceptions: bool = True, timeout_duration: int = 20) → List[Entry | PublisherFetchError | Exception][source]#

Fetch data from all fetchers asynchronously. Fetching is done in parallel for all fetchers.

Parameters:

filter_exceptions – If True, filters out exceptions from the result list
return_exceptions – If True, returns exceptions in the result list
timeout_duration – Timeout duration for each fetcher

Returns:

List of fetched data

fetch_sync(filter_exceptions: bool = True, return_exceptions: bool = True, timeout_duration: int = 20) → List[Entry | PublisherFetchError | Exception]#: Synchronous version of the method.

property fetchers: List[FetcherInterfaceT]#

HopHandler#

class pragma_sdk.common.fetchers.handlers.hop_handler.HopHandler[source]#

Bases: object

Dataclass in charge of handling pair hopping. Is mostly integrated within fetchers to handle different quote currencies.

Parameters:: hopped_currencies – Dict between the quote currency and the new quote currency

__init__(*args: Any, **kwargs: Any) → None#

get_hop_pair(pair: Pair) → Pair | None[source]#

Returns a new pair if the quote currency is in the hopped_currencies list Otherwise, returns None

Parameters:: pair – Pair
Returns:: Optional[Pair]

hopped_currencies: Dict[str, str]#

IndexAggregation#

class pragma_sdk.common.fetchers.handlers.index_aggregator_handler.AssetQuantities[source]#

Bases: object

__init__(pair: Pair, quantities: float)[source]#

class pragma_sdk.common.fetchers.handlers.index_aggregator_handler.IndexAggregatorHandler[source]#

Bases: object

__init__(spot_entries: List[SpotEntry], pair_quantities: List[AssetQuantities])[source]#

get_index_value()[source]#

pair_quantities: List[AssetQuantities]#

spot_entries: List[SpotEntry]#

standardize_decimals()[source]#

FetcherInterfaceT#

class pragma_sdk.common.fetchers.interface.FetcherInterfaceT[source]#

Bases: ABC

__init__(pairs: List[Pair], publisher: str, api_key: str | None = None, network: Annotated[Url, UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)] | Literal['devnet', 'mainnet', 'sepolia'] = 'mainnet')[source]#

abstract async fetch(session: ClientSession) → List[Entry | PublisherFetchError | BaseException][source]#: Fetches the data from the fetcher and returns a list of Entry objects.

abstract async fetch_pair(pair: Pair, session: ClientSession) → Entry | PublisherFetchError[source]#: Fetches the data for a specific pair from the fetcher and returns a SpotEntry object.

abstract fetch_pair_sync(pair: Pair, session: ClientSession) → Entry | PublisherFetchError#: Synchronous version of the method.

abstract fetch_sync(session: ClientSession) → List[Entry | PublisherFetchError | BaseException]#: Synchronous version of the method.

abstract format_url(pair: Pair) → str[source]#: Formats the URL for the fetcher, used in fetch_pair to get the data.

classmethod get_client(network: Annotated[Url, UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)] | Literal['devnet', 'mainnet', 'sepolia'] = 'mainnet') → PragmaOnChainClient[source]#

async get_stable_price(stable_asset: str) → float[source]#: Query the PragmaOnChainClient for the price of the stable asset in USD e.g get_stable_price(“USDT”) returns the price of USDT in USD

get_stable_price_sync(stable_asset: str) → float#: Synchronous version of the method.

headers: Dict[Any, Any]#

hop_handler: HopHandler | None = None#

pairs: List[Pair]#

publisher: str#