Source code for archivist.compliance

"""Compliance interface

   Access to the compliance 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") as tokenfile:
          authtoken =

      # Initialize connection to Archivist
      arch = Archivist(
      asset = arch.compliance.compliant_at(...)


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 .constants import (

LOGGER = getLogger(__name__)

[docs] class Compliance(dict): """Compliance Compliance object has dictionary of all the compliance attributes. """
# pylint: disable=too-few-public-methods
[docs] class _ComplianceClient: # pylint: disable=too-few-public-methods """ComplianceClient Access to compliance 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}/{COMPLIANCE_SUBPATH}" self._label = f"{self._subpath}/{COMPLIANCE_LABEL}" def __str__(self) -> str: return f"ComplianceClient({self._archivist.url})"
[docs] def compliant_at( self, asset_id, *, compliant_at: "bool|None" = None, report: "str|None" = None, ) -> Compliance: """ Reads compliance of a particular asset. Args: asset_id (str): asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxx compliant_at (str): datetime to check compliance at a particular time (optional). format: rfc3339 - UTC only report (bool): if true output report page_size (int): optional page size. (Rarely used). Returns: :class:`Compliance` instance """ params = {"compliant_at": compliant_at} if compliant_at is not None else None response = self._archivist.get( f"{self._label}/{asset_id}", params=params, ) if report is True: self.compliant_at_report(response) return Compliance(**response)
[docs] def compliant_at_report(self, compliance: "dict[str, Any]"): """ Prints report of compliance_at request Args: compliance (dict): compliance object encapsulating response from compliant_at """"Compliant %s", compliance["compliant"]) for outcome in compliance["compliance"]: if outcome["compliant"]: continue # get the compliance policy policy = outcome["compliance_policy_identity"] ) # print the policy name and the reason "NON-COMPLIANCE -> Policy: %s: Reason %s", policy["display_name"], outcome["reason"], )