Source code for mincepy.hist.metas

# -*- coding: utf-8 -*-
from typing import Any, Optional, Mapping, Dict, Iterator

from mincepy import archives
from mincepy import historians  # pylint: disable=unused-import, cyclic-import
from mincepy import exceptions

__all__ = ("Meta",)


[docs]class Meta: """A class for grouping metadata related methods""" # Meta is a 'friend' of Historian and so can access privates pylint: disable=protected-access def __init__(self, historian, archive): self._hist = historian # type: historians.Historian self._archive = archive self._sticky = {} @property def sticky(self) -> dict: return self._sticky
[docs] def get(self, obj_or_identifier) -> Optional[dict]: """Get the metadata for an object :param obj_or_identifier: either the object instance, an object ID or a snapshot reference """ results = self.get_many((obj_or_identifier,)) assert len(results) == 1 meta = tuple(results.values())[0] return meta
def get_many(self, obj_or_identifiers) -> Dict[Any, dict]: obj_ids = set(map(self._hist._ensure_obj_id, obj_or_identifiers)) trans = self._hist.current_transaction() if trans: # First, get what we can from the transaction found = {} for obj_id in obj_ids: try: found[obj_id] = trans.get_meta(obj_id) except exceptions.NotFound: pass # Now get anything else from the archive obj_ids -= found.keys() if obj_ids: from_archive = self._archive.meta_get_many(obj_ids) # Now put into the transaction so it doesn't look it up again. for obj_id in obj_ids: trans.set_meta(obj_id, from_archive[obj_id]) found.update(from_archive) return found # No transaction return self._archive.meta_get_many(obj_ids)
[docs] def set(self, obj_or_identifier, meta: Optional[Mapping]): """Set the metadata for an object :param obj_or_identifier: either the object instance, an object ID or a snapshot reference :param meta: the metadata dictionary """ obj_id = self._hist._ensure_obj_id(obj_or_identifier) trans = self._hist.current_transaction() if trans: return trans.set_meta(obj_id, meta) return self._archive.meta_set(obj_id, meta)
def set_many(self, metas: Mapping[Any, Optional[dict]]): mapped = { self._hist._ensure_obj_id(ident): meta for ident, meta in metas.items() } trans = self._hist.current_transaction() if trans: for entry in mapped.items(): trans.set_meta(*entry) else: self._archive.meta_set_many(mapped)
[docs] def update(self, obj_or_identifier, meta: Mapping): """Update the metadata for an object :param obj_or_identifier: either the object instance, an object ID or a snapshot reference :param meta: the metadata dictionary """ obj_id = self._hist._ensure_obj_id(obj_or_identifier) trans = self._hist.current_transaction() if trans: # Update the metadata in the transaction try: current = trans.get_meta(obj_id) except exceptions.NotFound: current = self._archive.meta_get(obj_id) # Try the archive if current is None: current = {} # Ok, no meta current.update(meta) trans.set_meta(obj_id, current) else: self._archive.meta_update(obj_id, meta)
def update_many(self, metas: Mapping[Any, Optional[dict]]): mapped = { self._hist._ensure_obj_id(ident): meta for ident, meta in metas.items() } trans = self._hist.current_transaction() if trans: for entry in mapped.items(): self.update(*entry) else: self._archive.meta_update_many(mapped)
[docs] def find( self, filter, obj_id=None ) -> Iterator[archives.Archive.MetaEntry]: # pylint: disable=redefined-builtin """Find metadata matching the given criteria. Each returned result is a tuple containing the corresponding object id and the metadata dictionary itself""" return self._archive.meta_find(filter=filter, obj_id=obj_id)
[docs] def distinct( self, key: str, filter: dict = None, # pylint: disable=redefined-builtin obj_id=None, ) -> Iterator: """Yield distinct values found for 'key' within metadata documents, optionally matching a search filter. The search can optionally be restricted to a set of passed object ids. :param key: the document key to get distinct values for :param filter: a query filter for the search :param obj_id: an optional restriction on the object ids to search. This ben be either: 1. a single object id 2. an iterable of object ids in which is treated as {'$in': list(obj_ids)} 3. a general query filter to be applied to the object ids """ return self._archive.meta_distinct(key, filter=filter, obj_id=obj_id)
[docs] def create_index(self, keys, unique=False, where_exist=False): """Create an index on the metadata. Takes either a single key or list of (key, direction) pairs :param keys: the key or keys to create the index on :param unique: if True, create a uniqueness constraint on this index :param where_exist: if True, only apply this index on documents that contain the key(s) """ self._archive.meta_create_index(keys, unique=unique, where_exist=where_exist)