##############################################################################
#
# Crossbar.io Database
# Copyright (c) typedef int GmbH. Licensed under MIT.
#
##############################################################################
import pprint
import uuid
import numpy as np
from zlmdb import MapUuidFlatBuffers, flatbuffers, table
from cfxdb import pack_uint256, unpack_uint256
from cfxdb.gen.xbrmm import Transaction as TransactionGen
[docs]
class _TransactionGen(TransactionGen.Transaction):
"""
Expand methods on the class code generated by flatc.
FIXME: come up with a PR for flatc to generated this stuff automatically.
"""
@classmethod
[docs]
def GetRootAsTransaction(cls, buf, offset):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = _TransactionGen()
x.Init(buf, n + offset)
return x
[docs]
def TidAsBytes(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
[docs]
def OfferAsBytes(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]
def AmountAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(14))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PaymentChannelAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(16))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PayingChannelAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(18))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def KeyAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(28))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def BuyerPubkeyAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(30))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PaymentChannelAfterAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(32))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PayingChannelAfterAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(34))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PaymentMmSigAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(36))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PaymentDelSigAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(38))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PayingMmSigAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(40))
if o != 0:
_off = self._tab.Vector(o)
_len = self._tab.VectorLen(o)
return memoryview(self._tab.Bytes)[_off : _off + _len]
return None
[docs]
def PayingDelSigAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(42))
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 Transaction(object):
"""
Data encryption key buy-sell transactions.
"""
"""
Unset
"""
"""
The transaction is currently in-flight
"""
"""
The transaction has completed with error (it failed)
"""
"""
The transaction has completed with success
"""
def __init__(self, from_fbs=None):
[docs]
self._from_fbs = from_fbs
# uint8[] (uuid)
# uint64
# uint32
[docs]
self._created_payment_channel_seq = None
# uint32
[docs]
self._created_paying_channel_seq = None
# uint8[] (uuid)
# uint8[] (uint256)
# uint8[] (uuid)
[docs]
self._payment_channel = None
# uint8[] (uuid)
[docs]
self._paying_channel = None
# uint8
# uint64
# uint32
[docs]
self._completed_payment_channel_seq = None
# uint32
[docs]
self._completed_paying_channel_seq = None
# [uint8] (uuid)
# [uint8] (uint256)
[docs]
self._buyer_pubkey = None
# [uint8] (uint256)
[docs]
self._payment_channel_after = None
# [uint8] (uint256)
[docs]
self._paying_channel_after = None
# [uint8] (ethsig)
[docs]
self._payment_mm_sig = None
# [uint8] (ethsig)
[docs]
self._payment_del_sig = None
# [uint8] (ethsig)
[docs]
self._paying_mm_sig = None
# [uint8] (ethsig)
[docs]
self._paying_del_sig = None
[docs]
def marshal(self):
obj = {
"tid": str(self.tid) if self.tid else None,
"created": self.created,
"created_payment_channel_seq": self._created_payment_channel_seq,
"created_paying_channel_seq": self._created_paying_channel_seq,
"offer": str(self.offer) if self.offer else None,
"amount": pack_uint256(self.amount) if self.amount else 0,
"payment_channel": self.payment_channel.bytes if self.payment_channel else None,
"paying_channel": self.paying_channel.bytes if self.paying_channel else None,
"state": self.state,
"completed": self.completed,
"completed_payment_channel_seq": self._completed_payment_channel_seq,
"completed_paying_channel_seq": self._completed_paying_channel_seq,
"key": self.key.bytes if self.key else None,
"buyer_pubkey": self.buyer_pubkey,
"payment_channel_after": pack_uint256(self.payment_channel_after) if self.payment_channel_after else None,
"paying_channel_after": pack_uint256(self.paying_channel_after) if self.paying_channel_after else None,
"payment_mm_sig": self.payment_mm_sig,
"payment_del_sig": self.payment_del_sig,
"paying_mm_sig": self.paying_mm_sig,
"paying_del_sig": self.paying_del_sig,
}
return obj
[docs]
def __str__(self):
return "\n{}\n".format(pprint.pformat(self.marshal()))
@property
[docs]
def tid(self) -> uuid.UUID:
"""
ID of the transaction.
"""
if self._tid is None and self._from_fbs:
if self._from_fbs.OfferLength():
_tid = self._from_fbs.TidAsBytes()
self._tid = uuid.UUID(bytes=bytes(_tid))
return self._tid
@tid.setter
def tid(self, value: uuid.UUID):
assert value is None or isinstance(value, uuid.UUID)
self._tid = value
@property
[docs]
def created(self) -> np.datetime64:
"""
Creation time of the transaction (epoch time in ns).
"""
if self._created is None and self._from_fbs:
self._created = np.datetime64(self._from_fbs.Created(), "ns")
return self._created
@created.setter
def created(self, value: np.datetime64):
assert value is None or isinstance(value, np.datetime64)
self._created = value
@property
[docs]
def created_payment_channel_seq(self) -> int:
"""
Sequence number of the created-state transaction within the payment channel.
"""
if self._created_payment_channel_seq is None and self._from_fbs:
self._created_payment_channel_seq = self._from_fbs.CreatedPaymentChannelSeq()
return self._created_payment_channel_seq or 0
@created_payment_channel_seq.setter
def created_payment_channel_seq(self, value: int):
assert value is None or type(value) == int
self._created_payment_channel_seq = value
@property
[docs]
def created_paying_channel_seq(self) -> int:
"""
Sequence number of the created-state transaction within the paying channel.
"""
if self._created_paying_channel_seq is None and self._from_fbs:
self._created_paying_channel_seq = self._from_fbs.CreatedPayingChannelSeq()
return self._created_paying_channel_seq or 0
@created_paying_channel_seq.setter
def created_paying_channel_seq(self, value: int):
assert value is None or type(value) == int
self._created_paying_channel_seq = value
@property
[docs]
def offer(self) -> uuid.UUID:
"""
ID of the data encryption key offer this transaction is for.
"""
if self._offer is None and self._from_fbs:
if self._from_fbs.OfferLength():
_offer = self._from_fbs.OfferAsBytes()
self._offer = uuid.UUID(bytes=bytes(_offer))
return self._offer
@offer.setter
def offer(self, value: uuid.UUID):
assert value is None or isinstance(value, uuid.UUID)
self._offer = value
@property
[docs]
def amount(self) -> int:
"""
Transaction amount in XBR.
"""
if self._amount is None and self._from_fbs:
if self._from_fbs.AmountLength():
_amount = self._from_fbs.AmountAsBytes()
self._amount = unpack_uint256(bytes(_amount))
else:
self._amount = 0
return self._amount
@amount.setter
def amount(self, value: int):
assert value is None or type(value) == int
self._amount = value
@property
[docs]
def payment_channel(self) -> uuid.UUID:
"""
Address of the payment channel (of the buyer) this transaction is transacting on.
"""
if self._payment_channel is None and self._from_fbs:
if self._from_fbs.PaymentChannelLength():
self._payment_channel = uuid.UUID(bytes=bytes(self._from_fbs.PaymentChannelAsBytes()))
return self._payment_channel
@payment_channel.setter
def payment_channel(self, value: uuid.UUID):
assert value is None or isinstance(value, uuid.UUID)
self._payment_channel = value
@property
[docs]
def paying_channel(self) -> uuid.UUID:
"""
Address of the paying channel (of the seller) this transaction is transacting on.
"""
if self._payment_channel is None and self._from_fbs:
if self._from_fbs.PayingChannelLength():
self._payment_channel = uuid.UUID(bytes=bytes(self._from_fbs.PayingChannelAsBytes()))
return self._payment_channel
@paying_channel.setter
def paying_channel(self, value: uuid.UUID):
assert value is None or isinstance(value, uuid.UUID)
self._payment_channel = value
@property
[docs]
def state(self) -> int:
"""
State of the transaction: TransactionState.INFLIGHT when created, and then either TransactionState.SUCCESS or TransactionState.FAILED.
"""
if self._state is None and self._from_fbs:
self._state = self._from_fbs.State()
return self._state or 0
@state.setter
def state(self, value: int):
assert value is None or type(value) == int
self._state = value
@property
[docs]
def completed(self) -> np.datetime64:
"""
Completion time of the transaction (epoch time in ns)
"""
if self._completed is None and self._from_fbs:
self._completed = np.datetime64(self._from_fbs.Completed(), "ns")
return self._completed
@completed.setter
def completed(self, value: np.datetime64):
assert value is None or isinstance(value, np.datetime64)
self._completed = value
@property
[docs]
def completed_payment_channel_seq(self) -> int:
"""
Sequence number of the completed-state transaction within the payment channel.
"""
if self._completed_payment_channel_seq is None and self._from_fbs:
self._completed_payment_channel_seq = self._from_fbs.CompletedPaymentChannelSeq()
return self._completed_payment_channel_seq or 0
@completed_payment_channel_seq.setter
def completed_payment_channel_seq(self, value: int):
assert value is None or type(value) == int
self._completed_payment_channel_seq = value
@property
[docs]
def completed_paying_channel_seq(self) -> int:
"""
Sequence number of the completed-state transaction within the paying channel.
"""
if self._completed_paying_channel_seq is None and self._from_fbs:
self._completed_paying_channel_seq = self._from_fbs.CompletedPayingChannelSeq()
return self._completed_paying_channel_seq or 0
@completed_paying_channel_seq.setter
def completed_paying_channel_seq(self, value: int):
assert value is None or type(value) == int
self._completed_paying_channel_seq = value
@property
[docs]
def key(self) -> uuid.UUID:
"""
ID of the data encryption key sold under the transaction.
"""
if self._key is None and self._from_fbs:
if self._from_fbs.KeyLength():
_key = self._from_fbs.KeyAsBytes()
self._key = uuid.UUID(bytes=bytes(_key))
return self._key
@key.setter
def key(self, value):
assert value is None or isinstance(value, uuid.UUID)
self._key = value
@property
[docs]
def buyer_pubkey(self) -> bytes:
"""
Buyer public key (Ed25519).
"""
if self._buyer_pubkey is None and self._from_fbs:
if self._from_fbs.BuyerPubkeyLength():
self._buyer_pubkey = self._from_fbs.BuyerPubkeyAsBytes()
return self._buyer_pubkey
@buyer_pubkey.setter
def buyer_pubkey(self, value: bytes):
assert value is None or (type(value) == bytes and len(value) == 32)
self._buyer_pubkey = value
@property
[docs]
def payment_channel_after(self) -> int:
"""
Payment channel balance after transaction.
"""
if self._payment_channel_after is None and self._from_fbs:
if self._from_fbs.PaymentChannelAfterLength():
_payment_channel_after = self._from_fbs.PaymentChannelAfterAsBytes()
self._payment_channel_after = unpack_uint256(bytes(_payment_channel_after))
else:
self._payment_channel_after = 0
return self._payment_channel_after
@payment_channel_after.setter
def payment_channel_after(self, value: int):
assert value is None or type(value) == int
self._payment_channel_after = value
@property
[docs]
def paying_channel_after(self) -> int:
"""
Payment channel balance after transaction.
"""
if self._paying_channel_after is None and self._from_fbs:
if self._from_fbs.PayingChannelAfterLength():
_paying_channel_after = self._from_fbs.PayingChannelAfterAsBytes()
self._paying_channel_after = unpack_uint256(bytes(_paying_channel_after))
else:
self._paying_channel_after = 0
return self._paying_channel_after
@paying_channel_after.setter
def paying_channel_after(self, value: int):
assert value is None or type(value) == int
self._paying_channel_after = value
@property
[docs]
def payment_mm_sig(self) -> bytes:
"""
Payment channel market maker transaction signature.
"""
if self._payment_mm_sig is None and self._from_fbs:
if self._from_fbs.PaymentMmSigLength():
self._payment_mm_sig = self._from_fbs.PaymentMmSigAsBytes()
return self._payment_mm_sig
@payment_mm_sig.setter
def payment_mm_sig(self, value: bytes):
assert value is None or (type(value) == bytes and len(value) == 65)
self._payment_mm_sig = value
@property
[docs]
def payment_del_sig(self) -> bytes:
"""
Payment channel (buyer) delegate transaction signature.
"""
if self._payment_del_sig is None and self._from_fbs:
if self._from_fbs.PaymentDelSigLength():
self._payment_del_sig = self._from_fbs.PaymentDelSigAsBytes()
return self._payment_del_sig
@payment_del_sig.setter
def payment_del_sig(self, value: bytes):
assert value is None or (type(value) == bytes and len(value) == 65)
self._payment_del_sig = value
@property
[docs]
def paying_mm_sig(self) -> bytes:
"""
Paying channel market maker transaction signature.
"""
if self._paying_mm_sig is None and self._from_fbs:
if self._from_fbs.PayingMmSigLength():
self._paying_mm_sig = self._from_fbs.PayingMmSigAsBytes()
return self._paying_mm_sig
@paying_mm_sig.setter
def paying_mm_sig(self, value: bytes):
assert value is None or (type(value) == bytes and len(value) == 65)
self._paying_mm_sig = value
@property
[docs]
def paying_del_sig(self) -> bytes:
"""
Paying channel (seller) delegate transaction signature.
"""
if self._paying_del_sig is None and self._from_fbs:
if self._from_fbs.PayingDelSigLength():
self._paying_del_sig = self._from_fbs.PayingDelSigAsBytes()
return self._paying_del_sig
@paying_del_sig.setter
def paying_del_sig(self, value: bytes):
assert value is None or (type(value) == bytes and len(value) == 65)
self._paying_del_sig = value
@staticmethod
[docs]
def cast(buf):
return Transaction(_TransactionGen.GetRootAsTransaction(buf, 0))
[docs]
def build(self, builder):
tid = self.tid.bytes if self.tid else None
if tid:
tid = builder.CreateString(tid)
offer = self.offer.bytes if self.offer else None
if offer:
offer = builder.CreateString(offer)
amount = self.amount
if amount:
amount = builder.CreateString(pack_uint256(amount))
payment_channel = self.payment_channel.bytes if self.payment_channel else None
if payment_channel:
payment_channel = builder.CreateString(payment_channel)
paying_channel = self.paying_channel.bytes if self.paying_channel else None
if paying_channel:
paying_channel = builder.CreateString(paying_channel)
key = self.key
if key:
key = builder.CreateString(key.bytes)
buyer_pubkey = self.buyer_pubkey
if buyer_pubkey:
buyer_pubkey = builder.CreateString(buyer_pubkey)
payment_channel_after = self.payment_channel_after
if payment_channel_after:
payment_channel_after = builder.CreateString(pack_uint256(payment_channel_after))
paying_channel_after = self.paying_channel_after
if paying_channel_after:
paying_channel_after = builder.CreateString(pack_uint256(paying_channel_after))
payment_mm_sig = self.payment_mm_sig
if payment_mm_sig:
payment_mm_sig = builder.CreateString(payment_mm_sig)
payment_del_sig = self.payment_del_sig
if payment_del_sig:
payment_del_sig = builder.CreateString(payment_del_sig)
paying_mm_sig = self.paying_mm_sig
if paying_mm_sig:
paying_mm_sig = builder.CreateString(paying_mm_sig)
paying_del_sig = self.paying_del_sig
if paying_del_sig:
paying_del_sig = builder.CreateString(paying_del_sig)
TransactionGen.TransactionStart(builder)
if tid:
TransactionGen.TransactionAddTid(builder, tid)
if self.created:
TransactionGen.TransactionAddCreated(builder, int(self.created))
if self.created_payment_channel_seq:
TransactionGen.TransactionAddCreatedPaymentChannelSeq(builder, self.created_payment_channel_seq)
if self.created_paying_channel_seq:
TransactionGen.TransactionAddCreatedPayingChannelSeq(builder, self.created_paying_channel_seq)
if offer:
TransactionGen.TransactionAddOffer(builder, offer)
if amount:
TransactionGen.TransactionAddAmount(builder, amount)
if payment_channel:
TransactionGen.TransactionAddPaymentChannel(builder, payment_channel)
if paying_channel:
TransactionGen.TransactionAddPayingChannel(builder, paying_channel)
if self.state:
TransactionGen.TransactionAddState(builder, int(self.state))
if self.completed:
TransactionGen.TransactionAddCompleted(builder, int(self.completed))
if self.completed_payment_channel_seq:
TransactionGen.TransactionAddCompletedPaymentChannelSeq(builder, self.completed_payment_channel_seq)
if self.completed_paying_channel_seq:
TransactionGen.TransactionAddCompletedPayingChannelSeq(builder, self.completed_paying_channel_seq)
if key:
TransactionGen.TransactionAddKey(builder, key)
if buyer_pubkey:
TransactionGen.TransactionAddBuyerPubkey(builder, buyer_pubkey)
if payment_channel_after:
TransactionGen.TransactionAddPaymentChannelAfter(builder, payment_channel_after)
if paying_channel_after:
TransactionGen.TransactionAddPayingChannelAfter(builder, paying_channel_after)
if payment_mm_sig:
TransactionGen.TransactionAddPaymentMmSig(builder, payment_mm_sig)
if payment_del_sig:
TransactionGen.TransactionAddPaymentDelSig(builder, payment_del_sig)
if paying_mm_sig:
TransactionGen.TransactionAddPayingMmSig(builder, paying_mm_sig)
if paying_del_sig:
TransactionGen.TransactionAddPayingDelSig(builder, paying_del_sig)
final = TransactionGen.TransactionEnd(builder)
return final
@table("87cf8eac-10d8-470d-a645-23e11343e065", build=Transaction.build, cast=Transaction.cast)
[docs]
class Transactions(MapUuidFlatBuffers):
"""
Data encryption key (off-chain) market transactions.
"""