LABLAB Custom Clients#

New in version 1.7.0


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).


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!


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


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).


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


Most implementations using these generators will match the following pattern:

next and .send essentially wrap the IO work in do_something here#
2    my_generator = self.generator(...)
4    values = next(my_generator)
5    new_values = do_something(values)
6    my_generator.send(new_values)
8except StopIteration as e:
9    return e.value




We instantiate our generator object


We produce a value, transform it, and send it back to the generator


Every thing is wrapped in a try-except to catch StopIteration


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.#
 2    my_generator = self.generator(...)
 4    values = next(my_generator)
 6    while True:
 7        new_values = do_something(values)
 8        values = my_generator.send(new_values)
10except StopIteration as e:
11    return e.value


Here is an example for implementing load from Subgrounds.load

def load(
    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)

        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})

    except StopIteration as e:
        return e.value

    assert False


Here is an example for implementing execute from Subgrounds.execute

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

      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`.

      A :class:`DataResponse` object representing the response

        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.