Source code for heedy.objects.objects

from typing import Dict

from ..base import APIObject, APIList, Session
from ..kv import KV

from .. import users
from .. import apps
from ..notifications import Notifications

from . import registry

[docs]class Object(APIObject): props = { "name", "description", "icon", "access", "meta", "tags", "key", "owner_scope", } def __init__(self, objectData: Dict, session: Session): super().__init__( f"api/objects/{objectData['id']}", {"object": objectData["id"]}, session, cached_data=objectData, ) # The object ID = objectData["id"] self._kv = KV(f"api/kv/objects/{objectData['id']}", self.session) @property def kv(self): return self._kv @kv.setter def kv(self, v): return self._kv.set(**v) def __getattr__(self, attr): try: return self.cached_data[attr] except: return None @property def owner(self): return users.User(self.cached_data["owner"], self.session) @property def app(self): if self.cached_data["app"] is None: return None return apps.App(self.cached_data["app"], session=self.session)
[docs] def update(self, **kwargs): """ Updates the given data:: o.update(name="My new name",description="my new description") """ meta = self.cached_data.get("meta", None) def updateMeta(o): if "result" in o and o["result"] == "ok" and "meta" in kwargs: if meta is None: self.cached_data.pop("meta", 0) else: # We have values of meta, so set them correctly kwm = kwargs["meta"] for key in kwm: if kwm[key] is None: meta.pop(key, 0) else: meta[key] = kwm[key] self.cached_data["meta"] = meta return o return self.session.f(super().update(**kwargs), updateMeta)
[docs]class Objects(APIList): def __init__(self, constraints: Dict, session: Session): super().__init__("api/objects", constraints, session) def __getitem__(self, item): return super()._getitem(item, f=lambda x: registry.getObject(x, self.session)) def __call__(self, **kwargs): # To query by object type, we use _type if "_type" in kwargs: kwargs["type"] = kwargs["_type"] del kwargs["_type"] return super()._call( f=lambda x: [registry.getObject(xx, self.session) for xx in x], **kwargs )
[docs] def create(self, name, meta={}, _type="timeseries", **kwargs): """ Creates a new object of the given type (timeseries by default). """ return super()._create( f=lambda x: registry.getObject(x, self.session), **{"name": name, "type": _type, "meta": meta, **kwargs}, )