Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:rtype: dict
.. code-block:: python
>>> from beem.account import Account
>>> account = Account("steemit")
>>> print(account.verify_account_authority(["STM7Q2rLBqzPzFeteQZewv9Lu3NLE69fZoLeL6YK59t7UmssCBNTU"])["valid"])
False
"""
if account is None:
account = self["name"]
elif isinstance(account, Account):
account = account["name"]
if not self.steem.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
if not isinstance(keys, list):
keys = [keys]
self.steem.rpc.set_next_node_on_empty_reply(False)
try:
if self.steem.rpc.get_use_appbase():
return self.steem.rpc.verify_account_authority({'account': account, 'signers': keys}, api="database")
else:
return self.steem.rpc.verify_account_authority(account, keys)
except MissingRequiredActiveAuthority:
return {'valid': False}
def unpin_post(self, account, permlink, mod_account):
""" Removes a post from the top of a community
:param str account: post author
:param str permlink: permlink
:param str mod_account: Account who broadcast this, (mods or higher)
"""
community = self["name"]
if not self.blockchain.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
json_body = [
'unpinPost', {
'community': community,
'account': account,
'permlink': permlink,
}
]
return self.blockchain.custom_json(
"community", json_body, required_posting_auths=[mod_account])
def _get_followers(self, direction="follower", last_user="", what="blog", limit=100):
""" Help function, used in get_followers and get_following
"""
if not self.steem.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
followers_list = []
limit_reached = True
cnt = 0
while limit_reached:
self.steem.rpc.set_next_node_on_empty_reply(False)
if self.steem.rpc.get_use_appbase():
query = {'account': self.name, 'start': last_user, 'type': what, 'limit': limit}
if direction == "follower":
followers = self.steem.rpc.get_followers(query, api='follow')['followers']
elif direction == "following":
followers = self.steem.rpc.get_following(query, api='follow')['following']
else:
if direction == "follower":
followers = self.steem.rpc.get_followers(self.name, last_user, what, limit, api='follow')
elif direction == "following":
followers = self.steem.rpc.get_following(self.name, last_user, what, limit, api='follow')
def get_potential_signatures(self):
""" Returns public key from signature
"""
if not self.blockchain.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
self.blockchain.rpc.set_next_node_on_empty_reply(False)
if self.blockchain.rpc.get_use_appbase():
args = {'trx': self.json()}
else:
args = self.json()
ret = self.blockchain.rpc.get_potential_signatures(args, api="database")
if 'keys' in ret:
ret = ret["keys"]
return ret
def _get_account_history(self, account=None, start=-1, limit=0):
if account is None:
account = self
account = Account(account, steem_instance=self.steem)
if not self.steem.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
self.steem.rpc.set_next_node_on_empty_reply(False)
if self.steem.rpc.get_use_appbase():
try:
ret = self.steem.rpc.get_account_history({'account': account["name"], 'start': start, 'limit': limit}, api="account_history")['history']
except ApiNotSupported:
ret = self.steem.rpc.get_account_history(account["name"], start, limit, api="condenser")
else:
ret = self.steem.rpc.get_account_history(account["name"], start, limit, api="database")
return ret
def flag_post(self, account, permlink, notes, reporter):
""" Suggest a post for the review queue
:param str account: post author
:param str permlink: permlink
:param str notes: notes
:param str reporter: Account who broadcast this
"""
community = self["name"]
if not self.blockchain.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
json_body = [
'flagPost', {
'community': community,
'account': account,
'permlink': permlink,
'notes': notes
}
]
return self.blockchain.custom_json(
"community", json_body, required_posting_auths=[reporter])
:rtype: list
.. code-block:: python
>>> from beem.account import Account
>>> account = Account("beem.app")
>>> account.get_account_votes()
[]
"""
if account is None:
account = self["name"]
elif isinstance(account, Account):
account = account["name"]
if not self.steem.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
self.steem.rpc.set_next_node_on_empty_reply(False)
if self.steem.rpc.get_use_appbase():
return self.steem.rpc.get_account_votes(account, api="condenser")
else:
return self.steem.rpc.get_account_votes(account)
def get_activities(self, limit=100, last_id=None):
""" Returns community activity
"""
community = self["name"]
if not self.blockchain.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
self.blockchain.rpc.set_next_node_on_empty_reply(False)
return self.blockchain.rpc.account_notifications({"account": community, "limit": limit, "last_id": last_id}, api="bridge")
def subscribe(self, account):
""" subscribe to a community
:param str account: account who suscribe to the community (is also broadcasting the custom_json)
"""
community = self["name"]
if not self.blockchain.is_connected():
raise OfflineHasNoRPCException("No RPC available in offline mode!")
json_body = [
'subscribe', {
'community': community,
}
]
return self.blockchain.custom_json(
"community", json_body, required_posting_auths=[account])