Predefined settings for any endpoint

Problem:
Creating assets usually requires some common settings and
it is tedious to repeatedly specify them when manipulating
entities.

Solution:
Add a fixtures property that allows the user to 'accumulate'
predefined fixed settings.

Signed-off-by: Paul Hewlett <phewlett76@gmail.com>

Author: eccles

README.rst

@@ -26,6 +26,7 @@ One can then use the examples code to create assets (see examples directory):
     
     from archivist.archivist import Archivist
     from archivist.errors import ArchivistError
+    from archivist.storage_integrity import StorageIntegrity
     
     # Oauth2 token that grants access
     with open(".auth_token", mode='r') as tokenfile:
@@ -54,9 +55,16 @@ One can then use the examples code to create assets (see examples directory):
             "some_custom_attribute": "value"  # You can add any custom value as long as
                                           # it does not start with arc_
     }
-    behaviours = ["Attachments", "RecordEvidence"]
-    
-    # The first argument is the behaviours of the asset
+    #
+    # store asset on the DLT or not. If DLT is not enabled for the user an error will occur if
+    # StorageIntegrity.LEDGER is specified. If unspecified then TENANT_STORAGE is used
+    # i.e. not stored on the DLT...
+    # storage_integrity = StorageIntegrity.TENANT_STORAGE
+    props = {
+        "storage_integrity": StorageIntegrity.LEDGER.name,
+    }
+
+    # The first argument is the properties of the asset
     # The second argument is the attributes of the asset
     # The third argument is wait for confirmation:
     #   If @confirm@ is True then this function will not
@@ -66,7 +74,7 @@ One can then use the examples code to create assets (see examples directory):
     #   it will be in the "Pending" status.
     #   Once it is added to the blockchain, the status will be changed to "Confirmed"
     try:
-        asset = arch.assets.create(behaviours, attrs=attrs, confirm=True)
+        asset = arch.assets.create(props=props, attrs=attrs, confirm=True)
     except Archivisterror as ex:
         print("error", ex)
     else:

archivist/access_policies.py

@@ -35,6 +35,7 @@
     ACCESS_POLICIES_LABEL,
     ASSETS_LABEL,
 )
+from .dictmerge import _deepmerge
 
 from .assets import Asset
 
@@ -43,6 +44,9 @@
 #: be changed.
 DEFAULT_PAGE_SIZE = 500
 
+FIXTURE_LABEL = "access_policies"
+
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -169,8 +173,8 @@ def delete(self, identity: str) -> Dict:
         """
         return self._archivist.delete(ACCESS_POLICIES_SUBPATH, identity)
 
-    @staticmethod
     def __query(
+        self,
         props: Optional[Dict],
         *,
         filters: Optional[List] = None,
@@ -183,7 +187,7 @@ def __query(
         if access_permissions is not None:
             query["access_permissions"] = access_permissions
 
-        return query
+        return _deepmerge(self._archivist.fixtures.get(FIXTURE_LABEL), query)
 
     def count(self, *, display_name: Optional[str] = None) -> int:
         """Count access policies.

archivist/archivist.py

@@ -38,7 +38,6 @@
 from collections import deque
 from requests.models import Response
 
-from flatten_dict import flatten
 import requests
 from requests_toolbelt.multipart.encoder import MultipartEncoder
 
@@ -48,6 +47,7 @@
     ROOT,
     SEP,
 )
+from .dictmerge import _deepmerge, _dotstring
 from .errors import (
     _parse_response,
     ArchivistBadFieldError,
@@ -105,6 +105,7 @@ def __init__(
         *,
         auth: Optional[str] = None,
         cert: Optional[str] = None,
+        fixtures: Optional[str] = None,
         verify: bool = True,
         max_time: int = MAX_TIME,
     ):
@@ -131,6 +132,7 @@ def __init__(
         self._response_ring_buffer = deque(maxlen=self.RING_BUFFER_MAX_LEN)
         self._session = requests.Session()
         self._max_time = max_time
+        self._fixtures = fixtures or {}
 
         # keep these in sync with CLIENTS map above
         self.assets: _AssetsClient
@@ -176,8 +178,17 @@ def cert(self) -> Optional[str]:
         """str: filepath containing authorisation certificate."""
         return self._cert
 
+    @property
+    def fixtures(self) -> Optional[dict]:
+        """dict: Contains predefined attributes for each endpoint"""
+        return self._fixtures
+
+    @fixtures.setter
+    def fixtures(self, fixtures: dict):
+        """dict: Contains predefined attributes for each endpoint"""
+        self._fixtures = _deepmerge(self._fixtures, fixtures)
+
     def __add_headers(self, headers: Optional[Dict]) -> Dict:
-        """docstring"""
         if headers is not None:
             newheaders = {**self.headers, **headers}
         else:
@@ -429,9 +440,7 @@ def last_response(self, *, responses: int = 1) -> List[Response]:
 
     @staticmethod
     def __query(query: Optional[Dict]):
-        return query and "&".join(
-            sorted(f"{k}={v}" for k, v in flatten(query, reducer="dot").items())
-        )
+        return query and "&".join(sorted(f"{k}={v}" for k, v in _dotstring(query)))
 
     def get_by_signature(
         self, path: str, field: str, query: Dict, *, headers: Optional[Dict] = None

archivist/assets.py

@@ -38,6 +38,7 @@
     CONFIRMATION_STATUS,
 )
 from . import confirmer
+from .dictmerge import _deepmerge
 from .errors import ArchivistNotFoundError
 
 #: Default page size - number of entities fetched in one REST GET in the
@@ -102,6 +103,9 @@ def name(self) -> NoneOnError[str]:
         return None
 
 
+FIXTURE_LABEL = "assets"
+
+
 class _AssetsClient:
     """AssetsClient
 
@@ -118,9 +122,9 @@ def __init__(self, archivist: "type_helper.Archivist"):
 
     def create(
         self,
-        props: Dict,
-        attrs: Dict,
         *,
+        props: Optional[Dict] = None,
+        attrs: Optional[Dict] = None,
         confirm: bool = False,
     ) -> Asset:
         """Create asset
@@ -196,13 +200,12 @@ def read(self, identity: str) -> Asset:
         """
         return Asset(**self._archivist.get(ASSETS_SUBPATH, identity))
 
-    @staticmethod
-    def __query(props: Optional[Dict], attrs: Optional[Dict]) -> Dict:
+    def __query(self, props: Optional[Dict], attrs: Optional[Dict]) -> Dict:
         query = deepcopy(props) if props else {}
         if attrs:
             query["attributes"] = attrs
 
-        return query
+        return _deepmerge(self._archivist.fixtures.get(FIXTURE_LABEL), query)
 
     def count(
         self, *, props: Optional[Dict] = None, attrs: Optional[Dict] = None

archivist/dictmerge.py

@@ -0,0 +1,22 @@
+"""Archivist dict deep merge
+"""
+
+from copy import deepcopy
+
+from flatten_dict import flatten, unflatten
+
+
+def _deepmerge(dct1: dict, dct2: dict) -> dict:
+    """Deep merge 2 dictionaries
+
+    The settings from dct2 overwrite or add to dct1
+    """
+    if dct1 is None:
+        return deepcopy(dct2)
+
+    return unflatten({**flatten(dct1), **flatten(dct2)})
+
+
+def _dotstring(dct: dict) -> str:
+    """Emit nested dictionary as dot delimited string"""
+    return flatten(dct, reducer="dot").items()

archivist/events.py

@@ -38,6 +38,7 @@
     EVENTS_LABEL,
 )
 from . import confirmer
+from .dictmerge import _deepmerge
 from .errors import ArchivistNotFoundError
 
 
@@ -46,6 +47,8 @@
 #: be changed.
 DEFAULT_PAGE_SIZE = 500
 
+FIXTURE_LABEL = "events"
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -211,17 +214,16 @@ def read(self, identity: str) -> Event:
             )
         )
 
-    @staticmethod
     def __query(
-        props: Optional[Dict], attrs: Optional[Dict], asset_attrs: Optional[Dict]
+        self, props: Optional[Dict], attrs: Optional[Dict], asset_attrs: Optional[Dict]
     ) -> Dict:
         query = deepcopy(props) if props else {}
         if attrs:
             query["event_attributes"] = attrs
         if asset_attrs:
             query["asset_attributes"] = asset_attrs
 
-        return query
+        return _deepmerge(self._archivist.fixtures.get(FIXTURE_LABEL), query)
 
     def count(
         self,

archivist/locations.py

@@ -30,13 +30,17 @@
 from archivist import archivist as type_helper
 
 from .constants import LOCATIONS_SUBPATH, LOCATIONS_LABEL
+from .dictmerge import _deepmerge
 
 
 #: Default page size - number of entities fetched in one REST GET in the
 #: :func:`~_LocationsClient.list` method. This can be overridden but should rarely
 #: be changed.
 DEFAULT_PAGE_SIZE = 500
 
+FIXTURE_LABEL = "locations"
+
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -117,13 +121,12 @@ def read(self, identity: str) -> Location:
             )
         )
 
-    @staticmethod
-    def __query(props: Optional[Dict], attrs: Optional[Dict]) -> Dict:
+    def __query(self, props: Optional[Dict], attrs: Optional[Dict]) -> Dict:
         query = props or {}
         if attrs:
             query["attributes"] = attrs
 
-        return query
+        return _deepmerge(self._archivist.fixtures.get(FIXTURE_LABEL), query)
 
     def count(
         self, *, props: Optional[Dict] = None, attrs: Optional[Dict] = None

archivist/subjects.py

@@ -32,12 +32,16 @@
     SUBJECTS_SUBPATH,
     SUBJECTS_LABEL,
 )
+from .dictmerge import _deepmerge
 
 #: Default page size - number of entities fetched in one REST GET in the
 #: :func:`~_SubjectsClient.list` method. This can be overridden but should rarely
 #: be changed.
 DEFAULT_PAGE_SIZE = 500
 
+FIXTURE_LABEL = "subjects"
+
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -171,8 +175,8 @@ def delete(self, identity: str) -> Dict:
         """
         return self._archivist.delete(SUBJECTS_SUBPATH, identity)
 
-    @staticmethod
     def __query(
+        self,
         *,
         display_name: Optional[str] = None,
         wallet_pub_keys: Optional[List[str]] = None,
@@ -190,7 +194,7 @@ def __query(
         if tessera_pub_keys is not None:
             query["tessera_pub_key"] = tessera_pub_keys
 
-        return query
+        return _deepmerge(self._archivist.fixtures.get(FIXTURE_LABEL), query)
 
     def count(self, *, display_name: Optional[str] = None) -> int:
         """Count subjects.

docs/features.rst

@@ -25,6 +25,7 @@ REST api (in any language):
        certain criteria to become confirmed.
     *  a **read_by_signature()** method that allows one to retrieve an asset or event with a 
        unique signature without knowing the identity.
+    *  predefined **fixtures** that allow specifying common attributes of assets/events
     *  comprehensive exception handling - clear specific exceptions.
     *  easily extensible - obeys the open-closed principle of SOLID where new endpoints 
        can be implemented by **extending** the package as opposed to modifying it.

docs/fixtures.rst

@@ -0,0 +1,66 @@
+.. _fixtures:
+
+Fixtures
+=============================================
+
+One can specify common attributes when creating/counting/querying assets, events
+and locations.
+
+.. code-block:: python
+    
+    from copy import deepcopy
+
+    from archivist.archivist import Archivist
+    from archivist.errors import ArchivistError
+    from archivist.storage_integrity import StorageIntegrity
+    
+    # Oauth2 token that grants access
+    with open(".auth_token", mode='r') as tokenfile:
+        authtoken = tokenfile.read().strip()
+    
+    # Initialize connection to Archivist - for assets on DLT.
+    ledger = Archivist(
+        "https://app.rkvst.io",
+        auth=authtoken,
+        fixtures = {
+            "assets": {
+                "storage_integrity": StorageIntegrity.LEDGER.name,
+            }
+        },
+    )
+    
+    # lets define doors in our namespace that reside on the ledger...
+    doors = deepcopy(ledger)
+    doors.fixtures = {
+        "assets": {
+            "attributes": {
+                "arc_display_type": "door",
+                "arc_namespace": "project xyz",
+            },
+        },
+    }
+
+    # a red front door
+    door = doors.assets.create(
+        attrs={
+            "arc_display_name": "front door",
+            "colour": "red",
+        },
+        confirm=True,
+    )
+
+    # a green back door
+    door = doors.assets.create(
+        attrs={
+            "arc_display_name": "back door",
+            "colour": "green",
+        },
+        confirm=True,
+    )
+
+    # no need to specify arc_display_type...
+    no_of_doors = doors.assets.count()
+    for d in doors.assets.list():
+        print(d)
+
+