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

184 lines
5.3 KiB
Python
Raw Normal View History

2021-02-01 18:12:51 +01:00
# standard imports
import datetime
import logging
# third-party imports
from sqlalchemy import Column, String, Integer, DateTime, ForeignKey
2021-03-24 15:52:52 +01:00
from chainlib.eth.constant import ZERO_ADDRESS
2021-02-01 18:12:51 +01:00
# local imports
from cic_eth.db.models.base import SessionBase
from cic_eth.db.models.tx import TxCache
from cic_eth.db.models.otx import Otx
logg = logging.getLogger()
class Lock(SessionBase):
"""Deactivate functionality on a global or per-account basis
"""
__tablename__ = "lock"
blockchain = Column(String)
address = Column(String, ForeignKey('tx_cache.sender'))
flags = Column(Integer)
date_created = Column(DateTime, default=datetime.datetime.utcnow)
otx_id = Column(Integer, ForeignKey('otx.id'))
def chain(self):
"""Get chain the cached instance represents.
"""
return self.blockchain
@staticmethod
2021-03-24 15:52:52 +01:00
def set(chain_str, flags, address=ZERO_ADDRESS, session=None, tx_hash=None):
2021-02-01 18:12:51 +01:00
"""Sets flags associated with the given address and chain.
If a flags entry does not exist it is created.
Does not validate the address against any other tables or components.
Valid flags can be found in cic_eth.db.enum.LockEnum.
:param chain_str: Chain spec string representation
:type str: str
:param flags: Flags to set
:type flags: number
:param address: Ethereum address
:type address: str, 0x-hex
:param session: Database session, if None a separate session will be used.
:type session: SQLAlchemy session
:returns: New flag state of entry
:rtype: number
"""
2021-03-01 21:15:17 +01:00
session = SessionBase.bind_session(session)
2021-02-01 18:12:51 +01:00
2021-03-01 21:15:17 +01:00
q = session.query(Lock)
2021-02-01 18:12:51 +01:00
#q = q.join(TxCache, isouter=True)
q = q.filter(Lock.address==address)
q = q.filter(Lock.blockchain==chain_str)
lock = q.first()
if lock == None:
lock = Lock()
lock.flags = 0
lock.address = address
lock.blockchain = chain_str
if tx_hash != None:
2021-03-01 21:15:17 +01:00
session.flush()
q = session.query(Otx)
2021-02-01 18:12:51 +01:00
q = q.filter(Otx.tx_hash==tx_hash)
otx = q.first()
if otx != None:
lock.otx_id = otx.id
lock.flags |= flags
r = lock.flags
2021-03-01 21:15:17 +01:00
session.add(lock)
session.commit()
SessionBase.release_session(session)
2021-02-01 18:12:51 +01:00
return r
@staticmethod
2021-03-24 15:52:52 +01:00
def reset(chain_str, flags, address=ZERO_ADDRESS, session=None):
2021-02-01 18:12:51 +01:00
"""Resets flags associated with the given address and chain.
If the resulting flags entry value is 0, the entry will be deleted.
Does not validate the address against any other tables or components.
Valid flags can be found in cic_eth.db.enum.LockEnum.
:param chain_str: Chain spec string representation
:type str: str
:param flags: Flags to set
:type flags: number
:param address: Ethereum address
:type address: str, 0x-hex
:param session: Database session, if None a separate session will be used.
:type session: SQLAlchemy session
:returns: New flag state of entry
:rtype: number
"""
2021-03-01 21:15:17 +01:00
session = SessionBase.bind_session(session)
2021-02-01 18:12:51 +01:00
2021-03-01 21:15:17 +01:00
q = session.query(Lock)
2021-02-01 18:12:51 +01:00
#q = q.join(TxCache, isouter=True)
q = q.filter(Lock.address==address)
q = q.filter(Lock.blockchain==chain_str)
lock = q.first()
r = 0
if lock != None:
lock.flags &= ~flags
if lock.flags == 0:
2021-03-01 21:15:17 +01:00
session.delete(lock)
2021-02-01 18:12:51 +01:00
else:
2021-03-01 21:15:17 +01:00
session.add(lock)
2021-02-01 18:12:51 +01:00
r = lock.flags
2021-03-01 21:15:17 +01:00
session.commit()
2021-02-01 18:12:51 +01:00
2021-03-01 21:15:17 +01:00
SessionBase.release_session(session)
2021-02-01 18:12:51 +01:00
return r
@staticmethod
2021-03-24 15:52:52 +01:00
def check(chain_str, flags, address=ZERO_ADDRESS, session=None):
2021-02-01 18:12:51 +01:00
"""Checks whether all given flags are set for given address and chain.
Does not validate the address against any other tables or components.
Valid flags can be found in cic_eth.db.enum.LockEnum.
:param chain_str: Chain spec string representation
:type str: str
:param flags: Flags to set
:type flags: number
:param address: Ethereum address
:type address: str, 0x-hex
:param session: Database session, if None a separate session will be used.
:type session: SQLAlchemy session
:returns: Returns the value of all flags matched
:rtype: number
"""
2021-03-01 21:15:17 +01:00
session = SessionBase.bind_session(session)
2021-02-01 18:12:51 +01:00
2021-03-01 21:15:17 +01:00
q = session.query(Lock)
2021-02-01 18:12:51 +01:00
#q = q.join(TxCache, isouter=True)
q = q.filter(Lock.address==address)
q = q.filter(Lock.blockchain==chain_str)
q = q.filter(Lock.flags.op('&')(flags)==flags)
lock = q.first()
r = 0
if lock != None:
r = lock.flags & flags
2021-03-01 21:15:17 +01:00
SessionBase.release_session(session)
2021-02-01 18:12:51 +01:00
return r
@staticmethod
def check_aggregate(chain_str, flags, address, session=None):
localsession = session
if localsession == None:
localsession = SessionBase.create_session()
r = Lock.check(chain_str, flags, session=localsession)
r |= Lock.check(chain_str, flags, address=address, session=localsession)
if session == None:
localsession.close()
return r