Source code for cfxdb.meta.attribute
##############################################################################
#
# Crossbar.io Database
# Copyright (c) typedef int GmbH. Licensed under MIT.
#
##############################################################################
import pprint
from uuid import UUID
import cbor2
import numpy as np
from zlmdb import MapUuidUuidStringFlatBuffers, flatbuffers, table
from cfxdb.gen.meta import Attribute as AttributeGen
[docs]
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
[docs]
def GetRootAsAttribute(cls, buf, offset):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = _AttributeGen()
x.Init(buf, n + offset)
return x
[docs]
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
[docs]
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
[docs]
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):
[docs]
self._from_fbs = from_fbs
# [uint8] (required, uuid)
# [uint8] (required, uuid)
[docs]
self._object_oid = None
# string (required)
# uint64 (timestamp)
# [uint8] (cbor)
[docs]
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
[docs]
def __str__(self):
return "\n{}\n".format(pprint.pformat(self.marshal()))
@property
[docs]
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
[docs]
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
[docs]
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
[docs]
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
[docs]
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
[docs]
def cast(buf):
return Attribute(_AttributeGen.GetRootAsAttribute(buf, 0))
[docs]
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
@table("42b1ca1f-f135-4761-8d10-f96a43612178", build=Attribute.build, cast=Attribute.cast)
[docs]
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`
"""