Module macrobond_data_api.web.subscription_list

Classes

class SubscriptionList (session: Session, last_modified: datetime.datetime, poll_interval: datetime.timedelta = None)

Class for polling and manipulating subscription lists. Subscription lists can be used to poll for updates at a specific frequency.

This class shouldn't be instantiated directly, but instead should be retrieved from a web client through subscription_list

Examples - for a program that polls for updates continuously

from datetime import datetime, timezone

with WebClient() as api:
    subscription_list = api.subscription_list(datetime.now(timezone.utc))
    subscription_list.set(['sek', 'nok'])
    while True:
        result = subscription_list.poll()
        for key, date in result.items():
            print(f'Series "{key}", last updated "{date}"')

Examples - for a program that retrieves all updates and then exits

from datetime import datetime, timezone

# this should be the last time you polled the subscription list.
last_modified = previous_last_modified # or datetime.now(timezone.utc) if it's the first time you poll.

with WebClient() as api:
    subscription_list = api.subscription_list(last_modified)
    for result in subscription_list.poll_until_no_more_changes():
        for key, date in result.items():
            print(f'Series "{key}", last updated "{date}"')

    # all done polling, so we can now update the last_modified time for next itme we poll.
    last_modified = subscription_list.last_modified

Browse git

Instance variables

var last_modified

Stores the date for when the subscription list was last modified.

var no_more_changes

An indicator that there are no changes at the moment.

var poll_interval

Specifies the time interval between polls.

Methods

def add(self, keys: Sequence[str]) ‑> None

Add one or more series to the subscription list.

Important: You should not make several calls in parallel that modifies the list.

Parameters

keys : Sequence[str]

A sequence of series names.

Browse git

def list(self) ‑> List[str]

Lists series currently registered in the subscription list.

Returns

List[str]

Browse git

def poll(self) ‑> Dict[str, datetime.datetime]

Polls for any changes on the series in the subscription list. If there are no updates, the method will return an empty dict after the poll interval time. This gives an opportunity to abort the polling loop.

Returns

Dict[str, datetime]]

A dictionary of names of series that have been updated, and the corresponding last update date.

Browse git

def poll_until_no_more_changes(self) ‑> Iterator[Dict[str, datetime.datetime]]

Polls for any changes on the series in the subscription list until there are no more changes.

Returns

Iterator[Dict[str, datetime]]]

An Iterator of dictionaries with names of series that have been updated, and the corresponding last update date.

Browse git

def remove(self, keys: Sequence[str]) ‑> None

Remove one or more series from the subscription list.

Important: You should not make several calls in parallel that modifies the list.

Parameters

keys : Sequence[str]

A sequence of series names.

Browse git

def set(self, keys: Sequence[str]) ‑> None

Set what series to include in the subscription list. This will replace all previous series in the list.

Important: You should not make several calls in parallel that modifies the list.

Parameters

keys : Sequence[str]

A sequence of series names.

Browse git