##############################################################################
#
# Crossbar.io Database
# Copyright (c) Crossbar.io Technologies GmbH. Licensed under MIT.
#
##############################################################################
import pprint
import flatbuffers
from cfxdb import unpack_uint256, pack_uint256
from cfxdb.gen.xbr import TokenApproval as TokenApprovalGen, TokenTransfer as TokenTransferGen
from zlmdb import table, MapBytes32FlatBuffers
class _TokenApprovalGen(TokenApprovalGen.TokenApproval):
"""
Expand methods on the class code generated by flatc.
FIXME: come up with a PR for flatc to generated this stuff automatically.
"""
@classmethod
def GetRootAsTokenApproval(cls, buf, offset):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = _TokenApprovalGen()
x.Init(buf, n + offset)
return x
def TxHashAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(4))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def BlockHashAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def OwnerAddressAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def SpenderAddressAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(10))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def ValueAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(12))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
[docs]class TokenApproval(object):
"""
``IERC20.TokenApproval`` event database object.
"""
def __init__(self, from_fbs=None):
self._from_fbs = from_fbs
self._tx_hash = None
self._block_hash = None
self._owner_address = None
self._spender_address = None
self._value = None
def marshal(self) -> dict:
obj = {
'tx_hash': bytes(self._tx_hash) if self._tx_hash else None,
'block_hash': bytes(self._block_hash) if self._block_hash else None,
'from_address': bytes(self._owner_address) if self._owner_address else None,
'to_address': bytes(self._spender_address) if self._spender_address else None,
'value': self._value,
}
return obj
def __str__(self):
return '\n{}\n'.format(pprint.pformat(self.marshal()))
@property
def tx_hash(self) -> bytes:
"""
Primary key: Transaction hash.
"""
if self._tx_hash is None and self._from_fbs:
if self._from_fbs.TxHashLength():
self._tx_hash = self._from_fbs.TxHashAsBytes()
return self._tx_hash
@tx_hash.setter
def tx_hash(self, value: bytes):
assert value is None or type(value) == bytes
self._tx_hash = value
@property
def block_hash(self) -> bytes:
"""
Block hash.
"""
if self._block_hash is None and self._from_fbs:
if self._from_fbs.TxHashLength():
self._block_hash = self._from_fbs.BlockHashAsBytes()
return self._block_hash
@block_hash.setter
def block_hash(self, value: bytes):
assert value is None or type(value) == bytes
self._block_hash = value
@property
def owner_address(self) -> bytes:
"""
XBR token sending address.
"""
if self._owner_address is None and self._from_fbs:
if self._from_fbs.OwnerAddressLength():
self._owner_address = self._from_fbs.OwnerAddressAsBytes()
return self._owner_address
@owner_address.setter
def owner_address(self, value: bytes):
assert value is None or type(value) == bytes
self._owner_address = value
@property
def spender_address(self) -> bytes:
"""
XBR token receiving address.
"""
if self._spender_address is None and self._from_fbs:
if self._from_fbs.TxHashLength():
self._spender_address = self._from_fbs.SpenderAddressAsBytes()
return self._spender_address
@spender_address.setter
def spender_address(self, value: bytes):
assert value is None or type(value) == bytes
self._spender_address = value
@property
def value(self) -> int:
"""
XBR token transferred.
"""
if self._value is None and self._from_fbs:
if self._from_fbs.ValueLength():
_value = self._from_fbs.ValueAsBytes()
self._value = unpack_uint256(bytes(_value))
else:
self._value = 0
return self._value
@value.setter
def value(self, value: int):
assert value is None or type(value) == int
self._value = value
@staticmethod
def cast(buf):
return TokenApproval(_TokenApprovalGen.GetRootAsTokenApproval(buf, 0))
def build(self, builder):
tx_hash = self.tx_hash
if tx_hash:
tx_hash = builder.CreateString(tx_hash)
block_hash = self.block_hash
if block_hash:
block_hash = builder.CreateString(block_hash)
owner_address = self.owner_address
if owner_address:
owner_address = builder.CreateString(owner_address)
spender_address = self.spender_address
if spender_address:
spender_address = builder.CreateString(spender_address)
value = self.value
if value:
value = builder.CreateString(pack_uint256(value))
TokenApprovalGen.TokenApprovalStart(builder)
if tx_hash:
TokenApprovalGen.TokenApprovalAddTxHash(builder, tx_hash)
if block_hash:
TokenApprovalGen.TokenApprovalAddBlockHash(builder, block_hash)
if owner_address:
TokenApprovalGen.TokenApprovalAddOwnerAddress(builder, owner_address)
if spender_address:
TokenApprovalGen.TokenApprovalAddSpenderAddress(builder, spender_address)
if value:
TokenApprovalGen.TokenApprovalAddValue(builder, value)
final = TokenApprovalGen.TokenApprovalEnd(builder)
return final
[docs]@table('f3d1f052-7160-48a7-b36a-221035fc9373', build=TokenApproval.build, cast=TokenApproval.cast)
class TokenApprovals(MapBytes32FlatBuffers):
"""
Persisted token approvals archive.
Map :class:`zlmdb.MapBytes32FlatBuffers` from ``transaction_hash`` to :class:`cfxdb.xbr.TokenApproval`
"""
class _TokenTransferGen(TokenTransferGen.TokenTransfer):
"""
Expand methods on the class code generated by flatc.
FIXME: come up with a PR for flatc to generated this stuff automatically.
"""
@classmethod
def GetRootAsTokenTransfer(cls, buf, offset):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = _TokenTransferGen()
x.Init(buf, n + offset)
return x
def TxHashAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(4))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def BlockHashAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def FromAddressAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def ToAddressAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(10))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
def ValueAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(12))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off:_off + _len]
return None
[docs]class TokenTransfer(object):
"""
``IERC20.TokenTransfer`` event database object.
"""
def __init__(self, from_fbs=None):
self._from_fbs = from_fbs
self._tx_hash = None
self._block_hash = None
self._from_address = None
self._to_address = None
self._value = None
def marshal(self) -> dict:
obj = {
'tx_hash': self._tx_hash,
'block_hash': self._block_hash,
'from_address': self._from_address,
'to_address': self._to_address,
'value': self._value,
}
return obj
def __str__(self):
return '\n{}\n'.format(pprint.pformat(self.marshal()))
@property
def tx_hash(self) -> bytes:
"""
Primary key: Transaction hash.
"""
if self._tx_hash is None and self._from_fbs:
if self._from_fbs.TxHashLength():
self._tx_hash = self._from_fbs.TxHashAsBytes()
return self._tx_hash
@tx_hash.setter
def tx_hash(self, value: bytes):
assert value is None or type(value) == bytes
self._tx_hash = value
@property
def block_hash(self) -> bytes:
"""
Block hash.
"""
if self._block_hash is None and self._from_fbs:
if self._from_fbs.TxHashLength():
self._block_hash = self._from_fbs.BlockHashAsBytes()
return self._block_hash
@block_hash.setter
def block_hash(self, value):
assert value is None or type(value) == bytes
self._block_hash = value
@property
def from_address(self) -> bytes:
"""
XBR token sending address.
"""
if self._from_address is None and self._from_fbs:
if self._from_fbs.TxHashLength():
self._from_address = self._from_fbs.FromAddressAsBytes()
return self._from_address
@from_address.setter
def from_address(self, value: bytes):
assert value is None or type(value) == bytes
self._from_address = value
@property
def to_address(self) -> bytes:
"""
XBR token receiving address.
"""
if self._to_address is None and self._from_fbs:
if self._from_fbs.TxHashLength():
self._to_address = self._from_fbs.ToAddressAsBytes()
return self._to_address
@to_address.setter
def to_address(self, value: bytes):
assert value is None or type(value) == bytes
self._to_address = value
@property
def value(self) -> int:
"""
XBR token transferred.
"""
if self._value is None and self._from_fbs:
if self._from_fbs.TxHashLength():
_value = self._from_fbs.ValueAsBytes()
self._value = unpack_uint256(bytes(_value))
else:
self._value = 0
return self._value
@value.setter
def value(self, value: int):
assert value is None or type(value) == int
self._value = value
@staticmethod
def cast(buf):
return TokenTransfer(_TokenTransferGen.GetRootAsTokenTransfer(buf, 0))
def build(self, builder):
tx_hash = self.tx_hash
if tx_hash:
tx_hash = builder.CreateString(tx_hash)
block_hash = self.block_hash
if block_hash:
block_hash = builder.CreateString(block_hash)
from_address = self.from_address
if from_address:
from_address = builder.CreateString(from_address)
to_address = self.to_address
if to_address:
to_address = builder.CreateString(to_address)
value = self.value
if value:
value = builder.CreateString(pack_uint256(value))
TokenTransferGen.TokenTransferStart(builder)
if tx_hash:
TokenTransferGen.TokenTransferAddTxHash(builder, tx_hash)
if block_hash:
TokenTransferGen.TokenTransferAddBlockHash(builder, block_hash)
if from_address:
TokenTransferGen.TokenTransferAddFromAddress(builder, from_address)
if to_address:
TokenTransferGen.TokenTransferAddToAddress(builder, to_address)
if value:
TokenTransferGen.TokenTransferAddValue(builder, value)
final = TokenTransferGen.TokenTransferEnd(builder)
return final
[docs]@table('7dcfffd2-4859-45d6-b9cd-ba7930a7a107', build=TokenTransfer.build, cast=TokenTransfer.cast)
class TokenTransfers(MapBytes32FlatBuffers):
"""
Persisted token transfers archive.
Map :class:`zlmdb.MapBytes32FlatBuffers` from ``transaction_hash`` to :class:`cfxdb.xbr.TokenTransfer`
"""