Source code for heedy.objects.objects

from typing import Dict
import pprint

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

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

from . import registry


[docs]class ObjectMeta: """ Heedy's objects have a metadata field which stores object type-specific information. This :code:`meta` property of an object is a key-value dictionary, and can be edited by altering the meta field in sync sessions, or by calling the :code:`update` method. .. tab:: Sync :: # A timeseries type object has a schema key in its metadata. Setting "schema" here # does not alter any other elements of :code:`meta`, but only the "schema" key. o.meta = {"schema": {"type": "string"}} .. tab:: Async :: # A timeseries type object has a schema key in its metadata. Setting "schema" here # does not alter any other elements of :code:`meta`, but only the "schema" key. await o.update(meta={"schema": {"type": "string"}}) The :code:`meta` property behaves like a dictionary, but has features that help in usage. For example, the above code can be written as: .. tab:: Sync :: o.meta.schema = {"type": "string"} .. tab:: Async :: await o.meta.update(schema={"type": "string"}) The :code:`meta` property also has several properties that offer syntactic sugar in synchronous code: .. tab:: Sync :: del o.meta.schema # Resets the schema to its default value del o.meta["schema"] # Same as above len(o.meta) # Returns the number of keys currently cached in the object metadata "schema" in o.meta # Returns True if the schema key is in the cached meta field .. tab:: Async :: await o.meta.delete("schema") # Resets the schema to its default value len(o.meta) # Returns the number of keys currently cached in the object metadata "schema" in o.meta # Returns True if the schema key is in the cached meta field Finally, the semantics of :code:`o.meta.schema` and :code:`o.meta["schema"]` are the same as for standard objects, meaning that :code:`o.meta["schema"]` does not query the server for the schema, but instead returns the cached values, while :code:`o.meta.schema` will always query the server, and needs to be awaited in async sessions. """ def __init__(self, obj): super().__setattr__("_object", obj) @property def cached_data(self): return self._object.cached_data["meta"]
[docs] def update(self, **kwargs): """Sets the given keys in the object's type metadata. .. tab:: Sync :: o.meta.update(schema={"type": "string"}) .. tab:: Async :: await o.meta.update(schema={"type": "string"}) Args: **kwargs: The keys to set and their values Returns: The updated object metadata (as a dictionary) Raises: HeedyException: If writing fails (usually due to insufficient permissions) """ return self._object.update(meta=kwargs)
[docs] def delete(self, *args: str): """Delete the given keys from the object metadata. Deleting a key resets the value of that property to its default. Removes the key from metadata if it is optional. .. tab:: Sync :: o.meta.delete("schema") assert o.meta["schema"] == {} .. tab:: Async :: await o.meta.delete("schema") assert o.meta["schema"] == {} Args: *args: The keys to delete Returns: The updated object metadata Raises: HeedyException: If writing fails (usually due to insufficient permissions) """ toDelete = {} for a in args: toDelete[a] = None return self._object.update(meta=toDelete)
def __call__(self): """ Queries the server for the object's metadata, and returns it as a dictionary. This does not need to be called if the object was already read, since an object read automatically caches its metadata. .. tab:: Sync :: metadict = o.meta() .. tab:: Async :: metadict = await o.meta() Usually, you will not need to call this method, since ObjectMeta behaves like a dictionary once the object is read. Returns: The object's metadata as a dictionary """ return self._object.session.f(self._object.read(), lambda o: o["meta"]) def __getattr__(self, attr): return self._object.session.f(self._object.read(), lambda o: o["meta"][attr]) def __getitem__(self, key: str): return self.cached_data[key] def __setitem__(self, name: str, value): return self._object.update(meta={name: value}) def __delitem__(self, name: str): return self.delete(name) def __setattr__(self, name: str, value): return self.__setitem__(name, value) def __delattr__(self, name: str): return self.__delitem__(name) def __iter__(self): return iter(self.cached_data) def __contains__(self, item): return item in self.cached_data def __len__(self): return len(self.cached_data) def __str__(self): return pprint.pformat(self.cached_data) def __repr__(self): return str(self) def __eq__(self, o: object) -> bool: if isinstance(o, ObjectMeta): return self._object == o._object else: return self.cached_data == o
[docs]class Object(APIObject): """ Object is the base class for all Heedy objects. For example, the Timeseries object type is a subclass of Object, and therefore includes all of the functionality described here. When an object of an unrecognized type is returned from the Heedy API, the Python client will return it as the Object type. """ props = { "name", "description", "icon", "access", "meta", "tags", "key", "owner_scope", } def __init__(self, cached_data: Dict, session: Session): super().__init__( f"api/objects/{q(cached_data['id'])}", {"object": cached_data["id"]}, session, cached_data=cached_data, ) self._kv = KV(f"api/kv/objects/{q(cached_data['id'])}", self.session) @property def id(self): """ The object's unique ID. This is directly available, so does not need to be awaited in async sessions:: print(myobj.id) """ return self.cached_data["id"] @property def kv(self): """ The key-value store associated with this object. For details of usage, see :ref:`python_kv`. Returns: A :class:`heedy.kv.KV` object for the element. """ return self._kv @kv.setter def kv(self, v): return self._kv.set(**v) @property def meta(self): """ The object type's metadata. For details of usage, see :ref:`python_objectmeta`. This does not need to be awaited in async sessions. Instead, to read the data if it was not yet cached, use :code:`o.meta()` or :code:`await o.meta()` depending on session type. Returns: An :class:`heedy.objects.objects.ObjectMeta` object for this element. See :ref:`python_objectmeta`. """ return ObjectMeta(self) @meta.setter def meta(self, v): return self.update(meta=v) def __getitem__(self, i): if i == "meta": return self.meta v = super().__getitem__(i) if i == "owner": return users.User(v, self.session) if i == "app" and v is not None: return apps.App(v, session=self.session) return v @property def owner(self): """ The user which owns this object: .. tab:: Sync :: print(o.owner.username) # Queries the server for the owner, and prints username print(o["owner"].username) # Uses cached query data to get the owner .. tab:: Async :: print((await o.owner).username) # Queries the server for the owner, and prints username print(o["owner"].username) # Uses cached query data to get the owner Returns: The :class:`~heedy.User` object of the user which owns this object. The returned user does not have any cached data other than its username. """ return self.session.f( self.read(), lambda x: users.User(x["owner"], self.session) ) @property def app(self): """ The app that manages this object, if any: .. tab:: Sync :: if obj.app is not None: # Query the server for the object's app print(obj["app"].id) # Use cached value from previous query .. tab:: Async :: if await obj.app is not None: # Query the server for the object's app print(obj["app"].id) # Use cached value from previous query When accessing the Heedy API as an app, you will not be able to see or access any other apps. Returns: The app that this object belongs to, or None if it does not belong to an app. The returned :class:`~heedy.App` object does not have any cached data other than its id, so you will need to call :code:`read()` on it if accessing more than its id. """ return self.session.f( self.read(), lambda x: apps.App(x["app"], session=self.session) if x["app"] is not None else None, )
[docs] def update(self, **kwargs): meta = self.cached_data.get("meta", {}) def updateMeta(o): if "result" in o and o["result"] == "ok" and "meta" in kwargs: if kwargs["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): """ Objects is a class implementing a list of objects. It is accessed as a property of users/apps/plugin to get the objects belonging to that user/app, or to query objects in all of heedy using plugin. .. tab:: Sync :: myuser.objects() # objects belonging to myuser myapp.objects() # objects managed by myapp plugin.objects() # all objects in heedy .. tab:: Async :: await myuser.objects() # objects belonging to myuser await myapp.objects() # objects managed by myapp await plugin.objects() # all objects in heedy """ def __init__(self, constraints: Dict, session: Session): super().__init__("api/objects", constraints, session)
[docs] def __getitem__(self, objectId: str): """Gets an object by its ID. Each object in heedy has a unique string ID, which can then be used to access the object. The ID can be seen in the URL of the object's page in the frontend. .. tab:: Sync :: obj = p.objects["d233rk43o6kkle43kl"] .. tab:: Async :: obj = await p.objects["d233rk43o6kkle43kl"] Returns: The :class:`~heedy.Object` with the given ID (or promise for the object) Throws: HeedyException: If the object does not exist """ return super()._getitem( objectId, f=lambda x: registry.getObject(x, self.session) )
[docs] def __call__(self, **kwargs): """Gets the objects matching the given constraints. If used as a property of a user of an app, it will return only the objects belonging to that user/app (the user/app constraint is automatically added). .. tab:: Sync :: obj = p.objects(type="timeseries", key="mykey", tags="tag1 tag2", owner="myuser", app="") .. tab:: Async :: obj = await p.objects(type="timeseries", key="mykey", tags="tag1 tag2", owner="myuser", app="") Args: type (str): The type of the objects (like "timeseries") key (str): The unique app key of the object (each app can only have one object with a given key) tags (str): The tags that the object must have, separated by spaces owner (str): The owner username of the object (set automatically when accessed using the objects property of a user) app (str): The app ID that the object belongs to. Set to empty string for objects that don't belong to any app. (set automatically when accessing the objects property of an app) icon (bool,False): Whether to include the icons of the returned objects' data. Returns: A list of :class:`~heedy.Object` matching the given constraints. Throws: HeedyException: If the request fails. """ return super()._call( f=lambda x: [registry.getObject(xx, self.session) for xx in x], **kwargs )
[docs] def create(self, name: str, meta: Dict = {}, type: str = "timeseries", **kwargs): """ Creates a new object of the given type (timeseries by default). Only the first argument, the object name is required. .. tab:: Sync :: obj = app.objects.create("My Timeseries", description="This is my timeseries", icon="fas fa-chart-line", type="timeseries", meta={"schema":{"type":"number"}}, tags="myts mydata" key="myts") .. tab:: Async :: obj = await app.objects.create("My Timeseries", description="This is my timeseries", icon="fas fa-chart-line", type="timeseries", meta={"schema":{"type":"number"}}, tags="myts mydata" key="myts") When creating an object for an app, it is useful to give it a `key`. Keys are unique per-app, meaning that the app can have only one object with the given key. Args: name (str,""): The name of the object description (str,""): A description of the object icon (str,""): The icon of the object, either a base64 urlencoded image or fontawesome/material icon id. owner (str): The owner of the object (set automatically when accessed using the objects property of a user or app). app (str): The app ID that the object belongs to (set automatically when accessing the objects property of an app). meta (dict,{}): The metadata of the object. type (str,"timeseries"): The type of the object key (str): The unique per-app key of the object tags (str,""): The tags assigned to the object, separated by spaces owner_scope (str,"*"): The space-separated scopes to give the owner of the object if it is managed by an app. Returns: The newly created :class:`~heedy.Object`, with its data cached. Throws: HeedyException: if the object could not be created. """ return super()._create( f=lambda x: registry.getObject(x, self.session), **{"name": name, "type": type, "meta": meta, **kwargs}, )