Source code for stellar_base.address
# coding: utf-8
from .horizon import Horizon
from .keypair import Keypair
from .exceptions import AccountNotExistError, NotValidParamError, HorizonError
from .horizon import HORIZON_LIVE, HORIZON_TEST
[docs]class Address(object):
"""The :class:`Address` object, which represents an address (public key) on
Stellar's network.
An :class:`Address` is initialized via a public key string, or derived via
a secret seed. The network on which the account exists is also specified,
as it is used to verify and set attributes via connecting to Horizon. It
mostly exists as a helper class for Horizon operations on a given account
ID.
:param str address: The address string that represents this
:class:`Address`.
:param str secret: The secret seed string that is used to derive the
address for this :class:`Address`.
:param str network: The network to connect to for verifying and retrieving
additional attributes from. 'PUBLIC' is an alias for 'Public Global Stellar Network ; September 2015',
'TESTNET' is an alias for 'Test SDF Network ; September 2015'. Defaults to TESTNET.
:param str horizon_uri: The horizon url to use for
connecting to for additional information for the account to which this
address corresponds to.
"""
# TODO: Make network an enum
def __init__(self,
address=None,
secret=None,
network='TESTNET',
horizon_uri=None):
if secret:
self.address = Keypair.from_seed(secret).address().decode()
elif address:
self.address = Keypair.from_address(address).address().decode()
else:
raise ValueError('oops, need a stellar address or secret')
# self.secret = secret
if network.upper() != 'PUBLIC':
self.network = 'TESTNET'
else:
self.network = 'PUBLIC'
if horizon_uri:
self.horizon = Horizon(horizon_uri)
elif network.upper() == 'PUBLIC':
self.horizon = Horizon(HORIZON_LIVE)
else:
self.horizon = Horizon(HORIZON_TEST)
self.sequence = None
self.balances = None
self.paging_token = None
self.thresholds = None
self.flags = None
self.signers = None
self.data = None
self.inflation_destination = None
self.subentry_count = None
[docs] def get(self):
"""Retrieve the account data that corresponds to this :class:`Address`.
Retrieve the account data from Horizon for the account that corresponds
to this :class:`Address`. Attempt to retrieve the following attributes
from Horizon:
* Sequence Number
* Balances
* Paging Token
* Thresholds
* Flags
* Signers
* Data
* Inflation Destination
* Subentry Count
:raises AccountNotExistError: If the account does not exist, shown by a
404 response from a Horizon server.
:raises Exception: If any other problems come up, or if a network
connection happens.
"""
try:
acc = self.horizon.account(self.address)
if acc.get('sequence'):
self.sequence = acc.get('sequence')
self.balances = acc.get('balances')
self.paging_token = acc.get('paging_token')
self.thresholds = acc.get('thresholds')
self.flags = acc.get('flags')
self.signers = acc.get('signers')
self.data = acc.get('data')
self.inflation_destination = acc.get('inflation_destination')
self.subentry_count = acc.get('subentry_count')
except HorizonError as err:
raise AccountNotExistError(err.message['title'])
[docs] def payments(self, sse=False, **kwargs):
"""Retrieve the payments JSON from this instance's Horizon server.
Retrieve the payments JSON response for the account associated with
this :class:`Address`.
:param bool sse: Use the SSE client for connecting to Horizon.
"""
self.__check_params(kwargs)
return self.horizon.account_payments(
self.address, params=kwargs, sse=sse)
[docs] def offers(self, **kwargs):
"""Retrieve the offers JSON from this instance's Horizon server.
Retrieve the offers JSON response for the account associated with
this :class:`Address`.
:param bool sse: Use the SSE client for connecting to Horizon.
"""
self.__check_params(kwargs)
return self.horizon.account_offers(self.address, params=kwargs)
[docs] def transactions(self, sse=False, **kwargs):
"""Retrieve the transactions JSON from this instance's Horizon server.
Retrieve the transactions JSON response for the account associated with
this :class:`Address`.
:param bool sse: Use the SSE client for connecting to Horizon.
"""
self.__check_params(kwargs)
return self.horizon.account_transactions(
self.address, params=kwargs, sse=sse)
[docs] def operations(self, sse=False, **kwargs):
"""Retrieve the operations JSON from this instance's Horizon server.
Retrieve the operations JSON response for the account associated with
this :class:`Address`.
:param bool sse: Use the SSE client for connecting to Horizon.
"""
self.__check_params(kwargs)
return self.horizon.account_operations(
self.address, params=kwargs, sse=sse)
[docs] def trades(self, sse=False, **kwargs):
"""Retrieve the trades JSON from this instance's Horizon server.
Retrieve the trades JSON response for the account associated with
this :class:`Address`.
:param bool sse: Use the SSE client for connecting to Horizon.
"""
self.__check_params(kwargs)
return self.horizon.account_trades(
self.address, params=kwargs, sse=sse
)
[docs] def effects(self, sse=False, **kwargs):
"""Retrieve the effects JSON from this instance's Horizon server.
Retrieve the effects JSON response for the account associated with
this :class:`Address`.
:param bool sse: Use the SSE client for connecting to Horizon.
"""
self.__check_params(kwargs)
return self.horizon.account_effects(
self.address, params=kwargs, sse=sse)
# noinspection PyMethodMayBeStatic
def __check_params(self, data):
"""Check for appropriate keywords for a Horizon request method.
Check a dict of arguments to make sure that they only contain allowable
params for requests to Horizon, such as 'cursor', 'limit', and 'order'.
"""
params_allowed = {'cursor', 'limit', 'order'}
params = set(data.keys())
if params - params_allowed:
raise NotValidParamError('not valid params')