##############################################################################
#
# Crossbar.io Database
# Copyright (c) Crossbar.io Technologies GmbH. Licensed under MIT.
#
##############################################################################
import pprint
from uuid import UUID
import cbor2
import flatbuffers
import numpy as np
from cfxdb.gen.meta import Attribute as AttributeGen
from zlmdb import table, MapUuidUuidStringFlatBuffers
class _AttributeGen(AttributeGen.Attribute):
"""
Expand methods on the class code generated by flatc.
FIXME: come up with a PR for flatc to generated this stuff automatically.
"""
@classmethod
def GetRootAsAttribute(cls, buf, offset):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = _AttributeGen()
x.Init(buf, n + offset)
return x
def TableOidAsBytes(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 ObjectOidAsBytes(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 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 Attribute(object):
"""
Generic meta-data attributes that can be stored on objects in tables.
"""
def __init__(self, from_fbs=None):
self._from_fbs = from_fbs
# [uint8] (required, uuid)
self._table_oid = None
# [uint8] (required, uuid)
self._object_oid = None
# string (required)
self._attribute = None
# uint64 (timestamp)
self._modified = None
# [uint8] (cbor)
self._value = None
def marshal(self) -> dict:
obj = {
'table_oid': self.table_oid.bytes if self.object_oid else None,
'object_oid': self.object_oid.bytes if self.object_oid else None,
'attribute': self.attribute,
'modified': int(self.modified) if self.modified else None,
'value': bytes(self._value) if self._value else None,
}
return obj
def __str__(self):
return '\n{}\n'.format(pprint.pformat(self.marshal()))
@property
def table_oid(self) -> UUID:
"""
Table of the object holding the attribute.
"""
if self._table_oid is None and self._from_fbs:
if self._from_fbs.TableOidLength():
_table_oid = self._from_fbs.TableOidAsBytes()
self._table_oid = UUID(bytes=bytes(_table_oid))
return self._table_oid
@table_oid.setter
def table_oid(self, value: UUID):
assert value is None or isinstance(value, UUID)
self._table_oid = value
@property
def object_oid(self) -> UUID:
"""
Object (within the table) holding the attribute.
"""
if self._object_oid is None and self._from_fbs:
if self._from_fbs.ObjectOidLength():
_object_oid = self._from_fbs.ObjectOidAsBytes()
self._object_oid = UUID(bytes=bytes(_object_oid))
return self._object_oid
@object_oid.setter
def object_oid(self, value: UUID):
assert value is None or isinstance(value, UUID)
self._object_oid = value
@property
def attribute(self) -> str:
"""
Attribute name (part of the key).
"""
if self._attribute is None and self._from_fbs:
attribute = self._from_fbs.Attribute()
if attribute:
self._attribute = attribute.decode('utf8')
return self._attribute
@attribute.setter
def attribute(self, value: str):
assert value is None or type(value) == str
self._attribute = value
@property
def modified(self) -> np.datetime64:
"""
Timestamp when the attribute was last modified (or first created).
"""
if self._modified is None and self._from_fbs:
self._modified = np.datetime64(self._from_fbs.Modified(), 'ns')
return self._modified
@modified.setter
def modified(self, value: np.datetime64):
assert value is None or isinstance(value, np.datetime64)
self._modified = value
@property
def value(self):
"""
Arbitrary attribute value, stored CBOR-serialized.
"""
if self._value is None and self._from_fbs:
if self._from_fbs.ValueLength():
_value = self._from_fbs.ValueAsBytes()
if _value:
self._value = cbor2.loads(_value)
return self._value
@value.setter
def value(self, new_value):
self._value = new_value
@staticmethod
def cast(buf):
return Attribute(_AttributeGen.GetRootAsAttribute(buf, 0))
def build(self, builder):
table_oid = self.table_oid.bytes if self.table_oid else None
if table_oid:
table_oid = builder.CreateString(table_oid)
object_oid = self.object_oid.bytes if self.object_oid else None
if object_oid:
object_oid = builder.CreateString(object_oid)
attribute = self.attribute
if attribute:
attribute = builder.CreateString(attribute)
value = self.value
if value:
value = builder.CreateString(cbor2.dumps(value))
AttributeGen.AttributeStart(builder)
if table_oid:
AttributeGen.AttributeAddTableOid(builder, table_oid)
if object_oid:
AttributeGen.AttributeAddObjectOid(builder, object_oid)
if attribute:
AttributeGen.AttributeAddAttribute(builder, attribute)
if self.modified:
AttributeGen.AttributeAddModified(builder, int(self.modified))
if value:
AttributeGen.AttributeAddValue(builder, value)
final = AttributeGen.AttributeEnd(builder)
return final
[docs]@table('42b1ca1f-f135-4761-8d10-f96a43612178', build=Attribute.build, cast=Attribute.cast)
class Attributes(MapUuidUuidStringFlatBuffers):
"""
Generic meta-data attributes that can be stored on objects in tables. Primary key of this table is ``(table_oid, object_oid, attribute)``.
Map :class:`zlmdb.MapUuidUuidStringFlatBuffers` from ``(table_oid, object_oid, attribute)`` to :class:`cfxdb.meta.Attribute`
"""