##############################################################################
#
# Crossbar.io Database
# Copyright (c) typedef int GmbH. Licensed under MIT.
#
##############################################################################
import pprint
import uuid
import numpy as np
from zlmdb import MapUuidFlatBuffers, MapUuidUuid, flatbuffers, table
from cfxdb import pack_uint256, unpack_uint256
from cfxdb.gen.xbrmm import Offer as OfferGen
[docs]
class _OfferGen(OfferGen.Offer):
"""
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 GetRootAsOffer(cls, buf, offset):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = _OfferGen()
x.Init(buf, n + offset)
return x
[docs]
def OfferAsBytes(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
[docs]
def SellerAsBytes(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
[docs]
def KeyAsBytes(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 ApiAsBytes(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 UriAsBytes(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 SignatureAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(22))
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 PriceAsBytes(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(24))
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 Offer(object):
"""
Data encryption key offerings by XBR providers.
"""
def __init__(self, from_fbs=None):
[docs]
self._from_fbs = from_fbs
# uint64
# [uint8] (uuid)
# [uint8] (address)
# uint64
[docs]
self._seller_session_id = None
# string
[docs]
self._seller_authid = None
# [uint8] (uuid)
# [uint8] (uuid)
# string
# uint64
[docs]
self._valid_from = None
# [uint8]
# [uint8] (uint256)
# [KeyValue]
[docs]
self._categories = None
# uint64
# uint32
# uint32
[docs]
self._remaining = None
[docs]
def marshal(self) -> dict:
obj = {
"timestamp": int(self.timestamp) if self.timestamp else None,
"offer": self.offer.bytes if self.offer else None,
"seller": bytes(self.seller) if self.seller else None,
"seller_session_id": self.seller_session_id,
"seller_authid": self.seller_authid,
"key": self.key.bytes if self.key else None,
"api": self.api.bytes if self.api else None,
"uri": self.uri,
"valid_from": int(self.valid_from) if self.valid_from else None,
"signature": bytes(self.signature) if self.signature else None,
"price": pack_uint256(self.price) if self.price else 0,
"categories": self.categories,
"expires": int(self.expires) if self.expires else None,
"copies": self.copies,
"remaining": self.remaining,
}
return obj
[docs]
def __str__(self):
return "\n{}\n".format(pprint.pformat(self.marshal()))
@property
[docs]
def timestamp(self) -> np.datetime64:
"""
Offer transaction time (epoch time in ns)
"""
if self._timestamp is None and self._from_fbs:
self._timestamp = np.datetime64(self._from_fbs.Timestamp(), "ns")
return self._timestamp
@timestamp.setter
def timestamp(self, value: np.datetime64):
assert value is None or isinstance(value, np.datetime64)
self._timestamp = value
@property
[docs]
def offer(self) -> uuid.UUID:
"""
ID of the data encryption key offer.
"""
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 seller(self) -> bytes:
"""
Address of the XBR provider offering the data encryption key.
"""
if self._seller is None and self._from_fbs:
if self._from_fbs.SellerLength():
self._seller = self._from_fbs.SellerAsBytes()
return self._seller
@seller.setter
def seller(self, value: bytes):
assert value is None or (type(value) == bytes and len(value) == 20)
self._seller = value
@property
[docs]
def seller_session_id(self) -> int:
"""
WAMP session ID of the caller that originally placed this offer.
"""
if self._seller_session_id is None and self._from_fbs:
self._seller_session_id = self._from_fbs.SellerSessionId()
return self._seller_session_id
@seller_session_id.setter
def seller_session_id(self, value: int):
assert value is None or type(value) == int
self._seller_session_id = value
@property
[docs]
def seller_authid(self) -> str:
"""
WAMP session authid of the caller that originally placed this offer.
"""
if self._seller_authid is None and self._from_fbs:
_seller_authid = self._from_fbs.SellerAuthid()
if _seller_authid:
self._seller_authid = _seller_authid.decode("utf8")
return self._seller_authid
@seller_authid.setter
def seller_authid(self, value):
self._seller_authid = value
@property
[docs]
def key(self) -> uuid.UUID:
"""
ID of the data encryption key offered.
"""
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 api(self) -> uuid.UUID:
"""
ID of the API the encrypted data (this key is for) is provided under.
:return:
"""
if self._api is None and self._from_fbs:
if self._from_fbs.ApiLength():
_api = self._from_fbs.ApiAsBytes()
self._api = uuid.UUID(bytes=bytes(_api))
return self._api
@api.setter
def api(self, value: uuid.UUID):
assert value is None or isinstance(value, uuid.UUID)
self._api = value
@property
[docs]
def uri(self) -> str:
"""
URI under which the data encrypted with the key offered is provided under.
"""
if self._uri is None and self._from_fbs:
_uri = self._from_fbs.Uri()
if _uri:
self._uri = _uri.decode("utf8")
return self._uri
@uri.setter
def uri(self, value: str):
self._uri = value
@property
[docs]
def valid_from(self) -> np.datetime64:
"""
Timestamp from which the offer is valid (epoch time in ns).
"""
if self._valid_from is None and self._from_fbs:
self._valid_from = np.datetime64(self._from_fbs.ValidFrom(), "ns")
return self._valid_from
@valid_from.setter
def valid_from(self, value: np.datetime64):
assert value is None or isinstance(value, np.datetime64)
self._valid_from = value
@property
[docs]
def signature(self) -> bytes:
"""
Seller delegate signature for the offer. The signature covers all information of the original offer placement request and requestor.
"""
if self._signature is None and self._from_fbs:
if self._from_fbs.SignatureLength():
self._signature = self._from_fbs.SignatureAsBytes()
return self._signature
@signature.setter
def signature(self, value: bytes):
assert value is None or type(value) == bytes
self._signature = value
@property
[docs]
def price(self) -> int:
"""
Price of data encryption key in XBR tokens.
"""
if self._price is None and self._from_fbs:
if self._from_fbs.PriceLength():
_price = self._from_fbs.PriceAsBytes()
self._price = unpack_uint256(bytes(_price))
else:
self._price = 0
return self._price
@price.setter
def price(self, value: int):
assert value is None or type(value) == int
self._price = value
@property
[docs]
def categories(self) -> dict:
"""
Dictionary of optional user defined categories the specific data that is provided falls under.
"""
if self._categories is None and self._from_fbs:
num = self._from_fbs.CategoriesKeyLength()
if num > 0:
categories = {}
for i in range(num):
key = self._from_fbs.CategoriesKey(i).decode("utf8")
value = self._from_fbs.CategoriesValue(i).decode("utf8")
categories[key] = value
self._categories = categories
return self._categories
@categories.setter
def categories(self, values: dict):
assert values is None or type(values) == dict
if values:
assert (type(key) == str for key in values.keys())
assert (type(value) == str for value in values.values())
self._categories = values
@property
[docs]
def expires(self) -> np.datetime64:
"""
Optional data at which this offer expires (epoch time in ns).
"""
if self._expires is None and self._from_fbs:
self._expires = np.datetime64(self._from_fbs.Expires(), "ns")
return self._expires
@expires.setter
def expires(self, value: np.datetime64):
assert value is None or isinstance(value, np.datetime64)
self._expires = value
@property
[docs]
def copies(self) -> int:
"""
Optional maximum number of times this data encryption key is to be sold or 0 for unlimited.
"""
if self._copies is None and self._from_fbs:
self._copies = self._from_fbs.Copies()
return self._copies
@copies.setter
def copies(self, value: int):
assert value is None or type(value) == int
self._copies = value
@property
[docs]
def remaining(self) -> int:
"""
Remaining number of copies to be sold (if "copies" is set >0, otherwise 0).
"""
if self._remaining is None and self._from_fbs:
self._remaining = self._from_fbs.Remaining()
return self._remaining
@remaining.setter
def remaining(self, value: int):
assert value is None or type(value) == int
self._remaining = value
@staticmethod
[docs]
def cast(buf):
return Offer(_OfferGen.GetRootAsOffer(buf, 0))
[docs]
def build(self, builder):
offer = self.offer.bytes if self.offer else None
if offer:
offer = builder.CreateString(offer)
seller = self.seller
if seller:
seller = builder.CreateString(seller)
seller_authid = self.seller_authid
if seller_authid:
seller_authid = builder.CreateString(seller_authid)
key = self.key.bytes if self.key else None
if key:
key = builder.CreateString(key)
api = self.api.bytes if self.api else None
if api:
api = builder.CreateString(api)
uri = self.uri
if uri:
uri = builder.CreateString(uri)
signature = self.signature
if signature:
signature = builder.CreateString(signature)
price = self.price
if price:
price = builder.CreateString(pack_uint256(price))
categories_keys_vec = None
categories_values_vec = None
if self._categories:
categories_keys = []
categories_values = []
for _key, _value in sorted(self._categories.items()):
assert type(_key) == str, "category key must be string, but was {}: {}".format(type(_key), _key)
assert type(_value) == str, "category value must be string, but was {}: {}".format(
type(_value), _value
)
categories_keys.append(builder.CreateString(_key))
categories_values.append(builder.CreateString(_value))
OfferGen.OfferStartCategoriesKeyVector(builder, len(categories_keys))
for _key in categories_keys:
builder.PrependUOffsetTRelative(_key)
categories_keys_vec = builder.EndVector()
OfferGen.OfferStartCategoriesValueVector(builder, len(categories_values))
for _value in categories_values:
builder.PrependUOffsetTRelative(_value)
categories_values_vec = builder.EndVector()
OfferGen.OfferStart(builder)
if self.timestamp:
OfferGen.OfferAddTimestamp(builder, int(self.timestamp))
if offer:
OfferGen.OfferAddOffer(builder, offer)
if seller:
OfferGen.OfferAddSeller(builder, seller)
if self.seller_session_id:
OfferGen.OfferAddSellerSessionId(builder, self.seller_session_id)
if seller_authid:
OfferGen.OfferAddSellerAuthid(builder, seller_authid)
if key:
OfferGen.OfferAddKey(builder, key)
if api:
OfferGen.OfferAddApi(builder, api)
if uri:
OfferGen.OfferAddUri(builder, uri)
if self.valid_from:
OfferGen.OfferAddValidFrom(builder, int(self.valid_from))
if signature:
OfferGen.OfferAddSignature(builder, signature)
if price:
OfferGen.OfferAddPrice(builder, price)
if categories_keys_vec:
OfferGen.OfferAddCategoriesKey(builder, categories_keys_vec)
if categories_values_vec:
OfferGen.OfferAddCategoriesValue(builder, categories_values_vec)
if self.expires:
OfferGen.OfferAddExpires(builder, int(self.expires))
if self.copies:
OfferGen.OfferAddCopies(builder, self.copies)
if self.remaining:
OfferGen.OfferAddRemaining(builder, self.remaining)
final = OfferGen.OfferEnd(builder)
return final
@table("dc6d175b-3dd0-4b1f-a6e8-2aec7f0e3fe5", build=Offer.build, cast=Offer.cast)
[docs]
class Offers(MapUuidFlatBuffers):
"""
Persisted data encryption key offers.
Map :class:`zlmdb.MapBytes32FlatBuffers` from ``offer_id`` to :class:`cfxdb.xbr.Offer`
"""
@table("ef5f1cdc-4871-4a03-ac1c-c60e80875b8b")
[docs]
class IndexOfferByKey(MapUuidUuid):
"""
Index: key_id -> offer_id
"""