LABLAB Custom Clients

New in version 1.7.0

Warning

This API is unstable and may change in the future!

Writing your own Subgrounds Client

Subclassing SubgroundsBase provides all the utlities for spinning up your own Subgrounds client. All of the business logic is handled, while you can define your own preferred API or bring your own IO solution (such as using requests or aiohttp instead of httpx).

Note

You can also choose to subclass Subgrounds or AsyncSubgrounds as they contain a more natural interface to work with, especially for simply swapping the implementation of the http client.

In the future, we may define a Protocol to help define a unified sync and async client structure!

Methodology

The SubgroundsBase class streamlines the all of the business logic neccessary to manage and make subgraph queries. Inheriting this class allows you to purely implement IO logic (making the actual request to the server). There's two main functions that allow you to

_load()

This deals with caching and loading subgraph schema data. If subgraph data is successfully loaded from a cache, the actual request execution will be skipped (via a StopIteration exception).

_execute()

This is the main entry point for executing queries via DataRequest and returning responses via DataResponse. Implementing execute allows you to

Both of these methods are implemented as generators as they provide us the utilities to lazily produce and consume values. The business logic produces objects to send to IO functions and then data is sent back into the business logic to continually transform until finally returned.

More on Generators

Wait, it's all generators?

Yes! Generators in Python are quite fancy. While they are generally seen as an easier way to create iterators, they also have the ability to consume values. Generators are the backbone for asyncio in Python as the keywords async and await are merely syntatical sugar for various forms of yield statements.

How are you using it?

We use generators via manually running next() and .send() functions to resume the generator state. Essentially, the business logic needs to step out and receive and send information to IO to compute. yield statements essentially allow us to "pause" the internal context while a different function implemented by a SubgroundsBase subclass to actually compute.

But why?

The approach, known as sans-io, we've chosen here has allowed us to reduce a large amount of code duplication when producing an async version of our API. It also disentangles the IO aspect of the library with the business logic allowing anyone to write in their own IO with very little work.

Where can I learn more?

• Basics on Sans-IO

• Generators as Pipelines

• Concurrency via Generators

• Generators beyond Iteration

Usage

Most implementations using these generators will match the following pattern:

next and .send essentially wrap the IO work in do_something here

try:
    my_generator = self.generator(...)

    values = next(my_generator)
    new_values = do_something(values)
    my_generator.send(new_values)

except StopIteration as e:
    return e.value

Step	Description
2	We instantiate our generator object
3-5	We produce a value, transform it, and send it back to the generator
1-8	Every thing is wrapped in a try-except to catch StopIteration
9	StopIteration.value is our actual return value!

Alternatively, you can host a while loop if you have an unknown number of producing and consuming to do:

Infinite generators, .send will also advance the generator like next.

try:
    my_generator = self.generator(...)

    values = next(my_generator)

    while True:
        new_values = do_something(values)
        values = my_generator.send(new_values)

except StopIteration as e:
    return e.value

load

Here is an example for implementing load from Subgrounds.load

def load(
    self,
    url: str,
    save_schema: bool = False,
    cache_dir: str | None = None,
    is_subgraph: bool = True,
) -> Subgraph:
    if cache_dir is not None:
        warnings.warn("This will be depreciated", DeprecationWarning)

    try:
        loader = self._load(url, save_schema, is_subgraph)
        url, query = next(loader)  # if this fails, schema is loaded from cache
        data = self._fetch(url, {"query": query})
        loader.send(data)

    except StopIteration as e:
        return e.value

    assert False

execute

Here is an example for implementing execute from Subgrounds.execute

def execute(
    self,
    req: DataRequest,
    pagination_strategy: Type[PaginationStrategy] | None = LegacyStrategy,
) -> DataResponse:
    """Executes a :class:`DataRequest` and returns a :class:`DataResponse`.

    Args:
      req: The :class:`DataRequest` object to be executed.
      pagination_strategy: A Class implementing the :class:`PaginationStrategy`
        ``Protocol``. If ``None``, then automatic pagination is disabled.
        Defaults to :class:`LegacyStrategy`.

    Returns:
      A :class:`DataResponse` object representing the response
    """

    try:
        executor = self._execute(req, pagination_strategy)

        doc = next(executor)
        while True:
            data = self._fetch(
                doc.url, {"query": doc.graphql, "variables": doc.variables}
            )
            doc = executor.send(DocumentResponse(url=doc.url, data=data))

    except StopIteration as e:
        return e.value

More on Sans-IO

Read forwards to get a bit more of an insight on our implementation of the sans-io technique.

• Sans-IO