Instanciating Clients

After registering resources in blacksmith, to consume API, a client must be instanciated. To create a client, service has to be discoverable using a service discovery strategy, then resources can be consumed.

Note

The service discovery part will be covered later in the document, this chapter is about consuming resources.

To create a client, a blacksmith.AsyncClientFactory object must be configured, instanciated, then will be responsible to build client for every registrated resources.

Synchronous code will use the blacksmith.SyncClientFactory instead.

from blacksmith import AsyncClientFactory, AsyncStaticDiscovery


async def main():
    sd = AsyncStaticDiscovery({("api", None): "http://srv:8000/"})
    cli = AsyncClientFactory(sd)
    api = await cli("api")
    result = await api.item.collection_get()
    if result.is_ok():
        for item in result.unwrap():
            print(item)
    else:
        print(result.unwrap_err())

In the example above, we consume the previously registered resource item, from the service api at version None.

The api.item property has methods collection_get, collection_post, get, patch, delete, to name a few to consume registered api routes. This section will be covered in the next section.

Important

Only registered routes works, consuming an unregistered route in the contract will raise error at the runtime. See Register Resources.

Type Hint

For a better development experience, type hints can be added, like the example bellow:

from result import Result

from blacksmith import AsyncClientFactory, AsyncStaticDiscovery, CollectionIterator

from .resources import Item, PartialItem


async def main():
    sd = AsyncStaticDiscovery({("api", None): "http://srv:8000/"})
    cli = AsyncClientFactory(sd)
    api = await cli("api")
    items: Result[CollectionIterator[PartialItem]] = await api.item.collection_get()
    for item in items.unwrap():
        full_item: Item = (await api.item.get({"name": item.name})).unwrap()
        print(full_item)

Note

methods that consume API such as .get(param) in the exemple above, accept both form get({"id": item.id}) and get(GetItem(id)) where GetItem is the Request Schema of the GET contract.

The method accept a dict version of the request schema.

Changed in version 2.0: Since blacksmith 2.0, responses are wrapped using the result library.

The collection class return a result.Result object, and the non collection, return a blacksmith.ResponseBox that have the same mimic of the result.Result. of the result library.

Synchronous API

Resource registration does not change for the sync/async version, but, all the runtime components differ. A prefix Async identified the asynchronous version and the prefix Sync define the synchronous version.

Lastly, here is the same example, using the synchronous API.

from result import Result

from blacksmith import CollectionIterator, SyncClientFactory, SyncStaticDiscovery

from .resources import Item, PartialItem


def main():
    sd = SyncStaticDiscovery({("api", None): "http://srv:8000/"})
    cli = SyncClientFactory(sd)
    api = cli("api")
    items: Result[CollectionIterator[PartialItem]] = api.item.collection_get()
    for item in items.unwrap():
        full_item: Item = (api.item.get({"name": item.name})).unwrap()
        print(full_item)