Golem Python API Reference

GolemNode

class yapapi.mid.golem_node.GolemNode(app_key=None, base_url=None)

Main entrypoint to the python Golem API, communicates with yagna.

GolemNode object corresponds to a single identity and a single running yagna instance (–> it’s identified by a (APPKEY, YAGNA_URL) pair) and can operate on different subnets / networks. Multiple GolemNode instances can be used to access different identities / yagna instances.

Usage:

golem = GolemNode()
async with golem:
    #   Interact with the Golem Network

__init__(app_key=None, base_url=None)

Parameters

app_key (Optional[str]) – App key used as an authentication token for all yagna calls. Defaults to the YAGNA_APPKEY env variable.
base_url (Optional[str]) – Base url for all yagna APIs. Defaults to YAGNA_API_URL env variable or http://127.0.0.1:7465.

async __aenter__()

Start. Initialize all the APIs and the event bus.

Return type: GolemNode

async __aexit__(*exc_info)

Shutdown. Stop collecting yagna events, close all resources created with autoclose=True, close APIs etc.

Return type: None

async create_allocation(amount, network='rinkeby', driver='erc20', autoclose=True)

Create a new allocation.

Parameters

amount (Union[Decimal, float]) – Amount of GLMs to be allocated
network (str) – Payment network
driver (str) – Payment driver
autoclose (bool) – Release allocation on __aexit__()

Return type

Allocation

async create_demand(payload, subnet='devnet-beta', expiration=None, allocations=(), autoclose=True, autostart=True)

Subscribe a new demand.

Parameters

payload (Payload) – Details of the demand
subnet (Optional[str]) – Subnet tag
expiration (Optional[datetime]) – Timestamp when all agreements based on this demand will expire (TODO: is this correct?)
allocations (Iterable[Allocation]) – Allocations that will be included in the description of this demand.
autoclose (bool) – Unsubscribe demand on __aexit__()
autostart (bool) – Immediately start collecting yagna events for this Demand. Without autostart events for this demand will start being collected after a call to start_collecting_events().

Return type

Demand

allocation(allocation_id)

Returns an Allocation with a given id (assumed to be correct, there is no validation).

Return type: Allocation

demand(demand_id)

Returns a Demand with a given id (assumed to be correct, there is no validation).

Return type: Demand

proposal(proposal_id, demand_id)

Returns a Proposal with a given id (assumed to be correct, there is no validation).

Id of a proposal has a meaning only in the context of a demand, so demand_id is also necessary (and also not validated).

Return type: Proposal

agreement(agreement_id)

Returns an Agreement with a given id (assumed to be correct, there is no validation).

Return type: Agreement

async allocations()

Returns a list of Allocation objects corresponding to all current allocations.

These are all allocations related to the current APP_KEY - it doesn’t matter if they were created with this GolemNode instance (or if GolemNode was used at all).

Return type: List[Allocation]

async demands()

Returns a list of Demand objects corresponding to all current demands.

These are all demands subscribed with the current APP_KEY - it doesn’t matter if they were created with this GolemNode instance (or if GolemNode was used at all).

Return type: List[Demand]

property event_bus: yapapi.mid.event_bus.EventBus

Returns the EventBus used by this GolemNode.

Any Event triggered by this GolemNode or any related object will be sent there and passed to registered listeners.

Return type: EventBus

High-Level API

[Nothing here yet. Task API, Service API etc.]

Mid-level API

Mid-level API consists of reusable components that can serve as a building blocks for various different applications.

Important temporary note: this will be easier to understand after reading the run.py example.

Chain

class yapapi.mid.chain.Chain(chain_start, *pipes)

Special class for mid-level components that utilize the pipes and filters pattern.

Sample usage:

async def source() -> AsyncIterator[int]:
    yield 1
    yield 2

async def int_2_str(numbers: AsyncIterator[int]) -> AsyncIterator[str]:
    async for number in numbers:
        yield str(number)

async for x in Chain(source, int_2_str):
    # x == "1"
    # x == "2"

More complex usages –> [TODO - examples, run.py].

Chain components

Components in this section can be used as parts of the Chain (but don’t have to).

class yapapi.mid.chain.SimpleScorer(score_proposal, min_proposals=None, max_wait=None)

Re-orders proposals using a provided scoring function.

__init__(score_proposal, min_proposals=None, max_wait=None)

Parameters

score_proposal (Callable[[Proposal], Awaitable[float]]) – Proposal-scoring function. Higher score -> better Proposal.
min_proposals (Optional[int]) – If not None, __call__() will not yield anything until at least that many proposals were scored (but max_wait overrides this)
max_wait (Optional[timedelta]) – If not None, we’ll not wait for min_proposals longer than that.

async __call__(proposals)

Consumes incoming proposals as fast as possible. Always yields a Proposal with the highest score.

Parameters: proposals (AsyncIterator[Proposal]) – Stream of Proposal to be reordered. In fact, this could be stream of whatever, as long as this whatever matches the scoring function, and this whatever would be yielded (–> TODO - maybe generalize this?).
Return type: AsyncIterator[Proposal]

class yapapi.mid.chain.DefaultNegotiator(buffer_size=1)

IN: any Proposal. OUT: a Proposal that can be used as a base for an Agreement

__init__(buffer_size=1)

Parameters: buffer_size (int) – (number_of_current_negotiations + number_of_ready_proposals) will always equal this number (except when the initial stream was exhausted, when it will be lower).

async __call__(proposals)

Consumes proposals as slowly as possible (but filling buffer_size), yields “ready” Proposal.

Parameters: proposals (AsyncIterator[Proposal]) – A stream of Proposal that will be used as a base of negotiations.
Return type: AsyncIterator[Proposal]

class yapapi.mid.chain.AgreementCreator

Uses a Proposal to create an Agreement, confirms it and waits for approval.

async __call__(proposals)

Consumes Proposal. Yields confirmed & approved Agreement.

At most a single Agreement is being created at any given time, so that’s pretty inefficient.

Parameters: proposals (AsyncIterator[Proposal]) – A stream of Proposal (that should be in a state that allows to Proposal.create_agreement on them).
Return type: AsyncIterator[Agreement]

Low-level API

Low-level objects correspond to resources in the Golem Network. They make no assumptions about any higher-level components that interact with them. Capabilities of the low-level API should match yagna capabilities, i.e. anything one can do by direct yagna interactions should also be possible - and, hopefully, more convenient - by performing operations on the low-level objects.

Resource

class yapapi.mid.resource.Resource(node, id_, data=None)

Base class of all low-level objects.

TODO - fix typing in the documentation of this class. I’m afraid this might require changes in sphinx_autodoc_typehints.

TODO - in the final version this should disappear from the public docs and all these methods should be described on subclasses, but this doesn’t make much sense before the previous TODO.

property parent: yapapi.mid.resource_internals.ParentType

Returns a Resource we are a child of. Details: children().

Return type: TypeVar(ParentType, _NULL, market.Proposal, Union[market.Demand, market.Proposal])

property children: List[yapapi.mid.resource_internals.ChildType]

List of Resources that were created “from” this resource.

E.g. children of a Demand are Proposal. Children of Proposal are either Proposal (counter-proposals to it), or Agreement.

Return type: List[TypeVar(ChildType, _NULL, market.Proposal, Union[market.Proposal, market.Agreement])]

async child_aiter()

Yields children. Stops when Resource knows there will be no more children.

Return type: AsyncIterator[TypeVar(ChildType, _NULL, market.Proposal, Union[market.Proposal, market.Agreement])]

property events: List[yapapi.mid.resource_internals.EventType]

Returns a list of all yagna events related to this Resource.

Note: these are yagna events and should not be confused with yapapi.mid.events.

Return type: List[TypeVar(EventType, _NULL, Union[ProposalEvent, ProposalRejectedEvent])]

property id: str

Id of the resource, generated by yagna.

Return type: str

property data: yapapi.mid.resource_internals.ModelType

Same as get_data(), but cached.

Raises RuntimeError if data was never fetched.

NOTE: data might be available even without a prior call to get_data() because some resources (e.g. initial Proposal) are fetched from yagna with full data right away.

Return type: TypeVar(ModelType, Allocation, Demand, Proposal, Agreement)

property node: yapapi.mid.golem_node.GolemNode

GolemNode that defines the context of this Resource

Return type: GolemNode

async get_data(force=False)

Returns details of this resource.

Parameters: force (bool) – False -> returns the cached data (or fetches data from yagna if there is no cached version). True -> always fetches the new data (and updates the cache).
Return type: TypeVar(ModelType, Allocation, Demand, Proposal, Agreement)

Market API

class yapapi.mid.market.Demand(node, id_, data=None)

A single demand on the Golem Network.

Created with one of the Demand-returning methods of the GolemNode.

start_collecting_events()

Start collecting yagna events in response to this demand.

Each event is either a new initial Proposal (yielded in initial_proposals()), a response to our counter-proposal (accessible via Proposal.responses()), or a rejection of a proposal.

Return type: None

async stop_collecting_events()

Stop collecting events, after a prior call to start_collecting_events().

Return type: None

async unsubscribe()

Stop all operations related to this demand and remove it.

This is a final operation, unsubscribed demand is not available anymore.

Return type: None

async initial_proposals()

Yields initial proposals matched to this demand.

Return type: AsyncIterator[Proposal]

proposal(proposal_id)

Return a Proposal with a given ID.

Return type: Proposal

class yapapi.mid.market.Proposal(node, id_, data=None)

A single proposal on the Golem Network.

Either a initial proposal matched to a demand, or a counter-proposal sent either by us or by the provider.

Sample usage:

initial_proposal = await demand.initial_proposals().__anext__()
our_counter_proposal = await initial_proposal.respond()
async for their_counter_proposal in our_counter_proposal.responses():
    agreement = their_counter_proposal.create_agreement()
    break
else:
    print("Our counter-proposal was rejected :(")

property initial: bool

True for proposals matched directly to the demand.

Return type: bool

property draft: bool

True for proposals that are responses to other proposals.

Return type: bool

property rejected: bool

True for rejected proposals. They will have no more responses().

Return type: bool

property demand: yapapi.mid.market.Demand

Initial Demand of this proposal.

Return type: Demand

async responses()

Yields responses to this proposal.

Stops when the proposal is rejected.

Return type: AsyncIterator[Proposal]

async create_agreement(autoclose=True, timeout=datetime.timedelta(seconds=60))

Promote this proposal to an agreement.

Parameters

autoclose (bool) – Terminate the agreement when the GolemNode closes.
timeout (timedelta) – TODO - this is used as AgreementValidTo, but what is it exactly?

Return type

Agreement

async reject(reason='')

Reject the proposal - inform the provider that we won’t send any more counter-proposals.

Parameters: reason (str) – An optional information for the provider describing rejection reasons.

Invalid on our responses.

Return type: None

async respond()

Respond to a proposal with a counter-proposal.

Invalid on proposals sent by the provider.

TODO: all the negotiation logic should be reflected in params of this method, but negotiations are not implemented yet.

Return type: Proposal

class yapapi.mid.market.Agreement(node, id_, data=None)

A single agreement on the Golem Network.

Sample usage:

agreement = await proposal.create_agreement()
await agreement.confirm()
await agreement.wait_for_approval()
#   Create activity, use the activity
await agreement.terminate()

async confirm()

Confirm the agreement.

First step that leads to an active agreement.

Return type: None

async wait_for_approval()

Wait for provider’s approval of the agreement.

Second (and last) step leading to an active agreement.

Return type: bool
Returns: True if agreement was approved.

async terminate(reason='')

Terminate the agreement.

Return type: None

Payment API

class yapapi.mid.payment.Allocation(node, id_, data=None)

A single allocation on the Golem Network.

Created with one of the Allocation-returning methods of the GolemNode.

async release()

Release the allocation.

Remaining amount will be available again. This is the final operation - released allocation is not available anymore.

Return type: None

Events

class yapapi.mid.event_bus.EventBus

Emit events, listen for events.

This class has few purposes:

Easy monitoring of the execution process (just log all events)
Convenient communication between separated parts of the code. E.g. we might want to act on incoming invoices from a component that is not connected to any other part of the code - with EventBus, we only have to register a callback for NewResource events.
Using a producer-consumer pattern to implement parts of the app-specific logic.

Sample usage:

async def on_allocation_event(event: ResourceEvent) -> None:
    print(f"Something happened to an allocation!", event)

golem = GolemNode()
event_bus: EventBus = golem.event_bus
event_bus.resource_listen(on_allocation_event, resource_classes=[Allocation])

async with golem:
    #   This will cause execution of on_allocation_event with a NewResource event
    allocation = await golem.create_allocation(1)
#   Allocation was created with autoclose=True, so now we also got a ResourceClosed event

listen(callback, classes=())

Execute the callback when Event is emitted.

Parameters

callback (Callable[[TypeVar(EventType, bound= Event)], Awaitable[None]]) – An async function to be executed.
classes (Iterable[Type[Event]]) – A list of Event subclasses - if not empty, callback will only be executed on events of matching classes.

Return type

None

resource_listen(callback, event_classes=(), resource_classes=(), ids=())

Execute the callback when ResourceEvent is emitted.

Parameters

callback (Callable[[TypeVar(ResourceEventType, bound= ResourceEvent)], Awaitable[None]]) – An async function to be executed.
event_classes (Iterable[Type[ResourceEvent]]) – A list of ResourceEvent subclasses - if not empty, callback will only be executed only on events of matching classes.
resource_classes (Iterable[Type[Resource]]) – A list of Resource subclasses - if not empty, callback will only be executed on events related to resources of a matching class.
ids (Iterable[str]) – A list of resource IDs - if not empty, callback will only be executed on events related to resources with a matching ID.

Return type

None

emit(event)

Emit an event - execute all callbacks listening for matching events.

If emit(X) was called before emit(Y), then it is guaranteed that callbacks for event Y will start only after all X callbacks finished (TODO - this should change, we don’t want the EventBus to stop because of a single never-ending callback).

Parameters: event (Event) – An event that will be emitted.
Return type: None

class yapapi.mid.events.Event: Base class for all events.

class yapapi.mid.events.ResourceEvent(resource)

Base class for all events related to a particular Resource.

property resource: yapapi.mid.resource.Resource

Resource related to this ResourceEvent.

Return type: Resource

class yapapi.mid.events.NewResource(resource)

Emitted when a new Resource object is created.

There are three distinct scenarios possible:

We create a new resource, e.g. with GolemNode.create_allocation()
We start interacting with some resource that was created by us before, but not with this particular GolemNode instance, eg. GolemNode.allocation()
We find a resource created by someone else (e.g. a Proposal)

There’s no difference between these scenarions from the POV of this event.

class yapapi.mid.events.ResourceDataChanged(resource, old_data)

Emitted when data attribute of a Resource changes.

This event is not emitted when the data “would have changed if we requested new data, but we didn’t request it”. In other words, we don’t listen for yagna-side changes, only react to changes already noticed in the context of a GolemNode.

Second argument is the old data (before the change), so comparing event.resource.data with event.old_data shows what changed.

NULL (i.e. empty) change doesn’t trigger the change, even if we explicitly sent a resource-changing call.

property old_data: Any

Value of self.resource.data before the change.

Return type: Any

class yapapi.mid.events.ResourceChangePossible(resource)

Emitted when we receive an information that a Resource.data might be outdated.

E.g. this is emitted for a Proposal when an agreement is created based on this proposal.

It is not guaranteed that anything really changed, but e.g. if you want to keep some Resource objects as up-to-date as possible, you might consider something like:

async def update_resource(event: ResourceEvent) -> None:
    await event.resource.get_data(force=True)

golem.event_bus.resource_listen(
    update_resource,
    event_classes=[ResourceChangePossible],
    ids=[an_important_resource.id]
)

NOTE: This is NOT IMPLEMENTED yet, as it’s pretty complex and useless now, and maybe not a good idea at all.

class yapapi.mid.events.ResourceClosed(resource)

Emitted when a resource is deleted or rendered unusable.

Usual case is when we delete a resource (e.g. Allocation.release()), or when the lifespan of a resource ends (e.g. Agreement.terminate()), but this can be also emitted when we notice that resource was deleted by someone else (TODO: when? is this possible at all? e.g. expired proposal?).

This is emitted only when something changes. Creating a new Resource object for an already closed resource (e.g. by passing an id of a terminated agreement to GolemNode.agreement) does not trigger this event.

This event should never be emitted more than once for a given Resource.

Logging

class yapapi.mid.default_logger.DefaultLogger(file_name='log.log')

Dump all events to a file.

Usage:

golem = GolemNode()
golem.event_bus.listen(DefaultLogger().on_event)

Or:

DefaultLogger().logger.debug("What's up?")

__init__(file_name='log.log')

Parameters: file_name (str) – Name of the file where all events will be dumped.

property file_name: str

Name of the file where all events will be dumped.

Return type: str

property logger: logging.Logger

Logger that just dumps everything to file file_name.

Return type: Logger

async on_event(event)

Callback that can be passed to EventBus.listen.

Return type: None