cic-internal-integration/apps/cic-eth/cic_eth/db/models/role.py

137 lines
3.3 KiB
Python
Raw Normal View History

2021-02-01 18:12:51 +01:00
# standard imports
import logging
# third-party imports
from sqlalchemy import Column, String, Text
from cic_registry import zero_address
# local imports
from .base import SessionBase
logg = logging.getLogger()
class AccountRole(SessionBase):
"""Key-value store providing plaintext tags for Ethereum addresses.
Address is initialized to the zero-address
:param tag: Tag
:type tag: str
"""
__tablename__ = 'account_role'
tag = Column(Text)
address_hex = Column(String(42))
2021-02-21 16:41:37 +01:00
# TODO:
2021-02-01 18:12:51 +01:00
@staticmethod
2021-02-21 16:41:37 +01:00
def get_address(tag, session):
2021-02-01 18:12:51 +01:00
"""Get Ethereum address matching the given tag
:param tag: Tag
:type tag: str
:returns: Ethereum address, or zero-address if tag does not exist
:rtype: str, 0x-hex
"""
2021-02-21 16:41:37 +01:00
if session == None:
raise ValueError('nested bind session calls will not succeed as the first call to release_session in the stack will leave the db object detached further down the stack. We will need additional reference count.')
session = SessionBase.bind_session(session)
role = AccountRole.__get_role(tag, session)
2021-02-21 16:41:37 +01:00
r = zero_address
if role != None:
r = role.address_hex
session.flush()
2021-02-21 16:41:37 +01:00
SessionBase.release_session(session)
return r
2021-02-01 18:12:51 +01:00
@staticmethod
2021-02-21 16:41:37 +01:00
def get_role(tag, session=None):
2021-02-01 18:12:51 +01:00
"""Get AccountRole model object matching the given tag
:param tag: Tag
:type tag: str
:returns: Role object, if found
:rtype: cic_eth.db.models.role.AccountRole
"""
2021-02-21 16:41:37 +01:00
session = SessionBase.bind_session(session)
role = AccountRole.__get_role(tag, session)
session.flush()
2021-02-21 16:41:37 +01:00
SessionBase.release_session(session)
2021-02-01 18:12:51 +01:00
return role
@staticmethod
2021-02-21 16:41:37 +01:00
def __get_role(tag, session):
q = session.query(AccountRole)
q = q.filter(AccountRole.tag==tag)
r = q.first()
return r
2021-02-01 18:12:51 +01:00
@staticmethod
2021-02-21 16:41:37 +01:00
def set(tag, address_hex, session=None):
2021-02-01 18:12:51 +01:00
"""Persist a tag to Ethereum address association.
This will silently overwrite the existing value.
:param tag: Tag
:type tag: str
:param address_hex: Ethereum address
:type address_hex: str, 0x-hex
:returns: Role object
:rtype: cic_eth.db.models.role.AccountRole
"""
2021-02-21 16:41:37 +01:00
session = SessionBase.bind_session(session)
role = AccountRole.__get_role(tag, session)
2021-02-01 18:12:51 +01:00
if role == None:
role = AccountRole(tag)
role.address_hex = address_hex
session.flush()
2021-02-21 16:41:37 +01:00
SessionBase.release_session(session)
return role
2021-02-01 18:12:51 +01:00
@staticmethod
def role_for(address, session=None):
"""Retrieve role for the given address
:param address: Ethereum address to match role for
:type address: str, 0x-hex
:returns: Role tag, or None if no match
:rtype: str or None
"""
2021-02-21 16:41:37 +01:00
session = SessionBase.bind_session(session)
2021-02-01 18:12:51 +01:00
2021-02-21 16:41:37 +01:00
q = session.query(AccountRole)
2021-02-01 18:12:51 +01:00
q = q.filter(AccountRole.address_hex==address)
role = q.first()
tag = None
if role != None:
tag = role.tag
2021-02-21 16:41:37 +01:00
SessionBase.release_session(session)
2021-02-01 18:12:51 +01:00
return tag
def __init__(self, tag):
self.tag = tag
self.address_hex = zero_address