Source code for archivist.locations

"""Locations interface

   Direct access to the locations endpoint.

   The user is not expected to use this class directly. It is an attribute of the
   :class:`Archivist` class.

   For example instantiate an Archivist instance and execute the methods of the class:

   .. code-block:: python

      with open(".auth_token", mode="r", encoding="utf-8") as tokenfile:
          authtoken =

      # Initialize connection to Archivist
      arch = Archivist(
      location = arch.locations.create(...)


from contextlib import suppress
from copy import deepcopy
from logging import getLogger
from typing import TYPE_CHECKING, Any

    # pylint:disable=cyclic-import      # but pylint doesn't understand this feature
    from .archivist import Archivist

from .dictmerge import _deepmerge
from .errors import ArchivistNotFoundError
from .utils import selector_signature

LOGGER = getLogger(__name__)

[docs] class Location(dict): """Location Location object has dictionary attributes. """ @property def name(self) -> "str | None": """str: name of the location""" name = None with suppress(KeyError): name = self["display_name"] return name
[docs] class _LocationsClient: """LocationsClient Access to locations entities using CRUD interface. This class is usually accessed as an attribute of the Archivist class. Args: archivist (Archivist): :class:`Archivist` instance """ def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{LOCATIONS_SUBPATH}" self._label = f"{self._subpath}/{LOCATIONS_LABEL}" def __str__(self) -> str: return f"LocationsClient({self._archivist.url})"
[docs] def create( self, props: "dict[str, Any]", *, attrs: "dict[str, Any]|None" = None ) -> Location: """Create location Creates location with defined properties and attributes. Args: props (dict): properties for this location. attrs (dict): attributes of created location. Returns: :class:`Location` instance """ LOGGER.debug("Create Location %s", props) return self.create_from_data(self.__params(props, attrs))
[docs] def create_from_data(self, data: "dict[str, Any]") -> Location: """Create location Creates location with request body from data stream. Suitable for reading data from a file using json.load or yaml.load Args: data (dict): request body of location. Returns: :class:`Location` instance """ return Location(**, data))
[docs] def create_if_not_exists(self, data: "dict[str, Any]") -> "tuple[Location, bool]": """ Create a location if not already exists Args: data (dict): request body of location. A YAML representation of the data argument would be: .. code-block:: yaml selector: - display_name - attributes: - wavestone_ext display_name: Apartements du Gare du Nord description: Residential apartment building in new complex above GdN station latitude: 48.8809 longitude: 2.3553 attributes: address: 18 Rue de Dunkerque, 75010 Paris, France wavestone_ext: managed The 'selector' setting is required. Returns: tuple of :class:`Location` instance, Boolean True if already exists """ data = deepcopy(data) selector = data.pop("selector") # must exist props, attrs = selector_signature(selector, data) try: location = self.read_by_signature(props=props, attrs=attrs) except ArchivistNotFoundError: "location with selector %s,%s does not exist - creating", props, attrs ) else:"location with selector %s,%s already exists", props, attrs) return location, True return self.create_from_data(data), False
[docs] def read(self, identity: str) -> Location: """Read location Reads location. Args: identity (str): location identity e.g. locations/xxxxxxxxxxxxxxxxxxxxxxx Returns: :class:`Location` instance """ return Location(**self._archivist.get(f"{self._subpath}/{identity}"))
def __params( self, props: "dict[str, Any]|None", attrs: "dict[str, Any]|None" ) -> "dict[str, Any]": params = deepcopy(props) if props else {} if attrs: params["attributes"] = attrs return _deepmerge(self._archivist.fixtures.get(LOCATIONS_LABEL), params)
[docs] def count( self, *, props: "dict[str, Any]|None" = None, attrs: "dict[str, Any]|None" = None, ) -> int: """Count locations. Counts number of locations that match criteria. Args: props (dict): e.g. {"display_name": "Macclesfield" } attrs (dict): e.g. {"director": "john smith" } Returns: integer count of locations. """ return self._archivist.count(self._label, params=self.__params(props, attrs))
[docs] def list( self, *, page_size: "int|None" = None, props: "dict[str, Any]|None" = None, attrs: "dict[str, Any]|None" = None, ): """List locations. Lists locations that match criteria. Args: props (dict): optional e.g. {"display_name": "Macclesfield" } attrs (dict): optional e.g. {"director": "john smith" } page_size (int): optional page size. (Rarely used) Returns: iterable that returns :class:`Location` instances """ return ( Location(**a) for a in self._archivist.list( self._label, LOCATIONS_LABEL, page_size=page_size, params=self.__params(props, attrs), ) )
[docs] def read_by_signature( self, *, props: "dict[str, Any]|None" = None, attrs: "dict[str, Any]|None" = None, ) -> Location: """Read location by signature. Reads location that meets criteria. Only one location is expected. Args: props (dict): e.g. {"display_name": "Macclesfield" } attrs (dict): e.g. {"director": "john smith" } Returns: :class:`Location` instance """ return Location( **self._archivist.get_by_signature( self._label, LOCATIONS_LABEL, params=self.__params(props, attrs), ) )