Merge pull request openwallet-foundation#23 from krzosa/master

Add linking to pds as an OYD extension. Add linking dris to service flow.

aries_cloudagent/aathcf/utils.py

@@ -1,5 +1,5 @@
 from aries_cloudagent.messaging.base_handler import HandlerException
-from aries_cloudagent.pdstorage_thcf.api import pds_get_active
+
 
 from aries_cloudagent.pdstorage_thcf.models.saved_personal_storage import SavedPDS
 from aries_cloudagent.config.settings import Settings
@@ -35,6 +35,8 @@ async def build_context(pds_type="own_your_data", connection_record={}):
     context = await build_standalone_context()
     default_storage = SavedPDS(type=pds_type, state=SavedPDS.ACTIVE)
     await default_storage.save(context)
+    from aries_cloudagent.pdstorage_thcf.api import pds_get_active
+
     pds = await pds_get_active(context)
 
     if pds_type == "own_your_data":

aries_cloudagent/pdstorage_thcf/api.py

@@ -1,13 +1,16 @@
+from aries_cloudagent.aathcf.utils import build_context, run_standalone_async
+from aries_cloudagent.aathcf.credentials import assert_type, assert_type_or
+from aries_cloudagent.storage.error import StorageNotFoundError
 from .base import BasePDS
-from .error import PDSNotFoundError
+from .error import PDSError, PDSNotFoundError, PDSRecordNotFoundError
 from .models.saved_personal_storage import SavedPDS
 import hashlib
 import multihash
 import logging
 import multibase
-from aries_cloudagent.storage.error import StorageNotFoundError
+
 from .models.table_that_matches_dris_with_pds import DriStorageMatchTable
-from aries_cloudagent.aathcf.credentials import assert_type, assert_type_or
+
 import json
 from collections import OrderedDict
 
@@ -77,6 +80,60 @@ async def pds_load(context, id: str, *, with_meta: bool = False) -> dict:
         return result["content"]
 
 
+async def pds_link_dri(context, dri, link_with_dris):
+    if isinstance(link_with_dris, str):
+        link_with_dris = [link_with_dris]
+    elif isinstance(link_with_dris, list):
+        pass
+    else:
+        raise TypeError("Expected: list or string")
+
+    if __debug__:
+        assert isinstance(link_with_dris, list)
+
+    pds = await pds_get_active(context)
+    if "link" in dir(pds):
+        await pds.link(dri, link_with_dris)
+    else:
+        LOGGER.warning(
+            "This is an own your data pds extension. current pds doesn't support this method"
+        )
+
+
+async def __test_pds_link():
+    context = await build_context()
+
+    import random
+
+    dri1 = await pds_save_a(context, "124inmjfa0dioq-0dq" + str(random.random()))
+    dri2 = await pds_save_a(context, "1asdad24inmjfa0dioq-0dq" + str(random.random()))
+    await pds_link_dri(context, dri1, dri2)
+
+    try:
+        await pds_link_dri(context, "1wd1fc", "difhd908f")
+    except PDSRecordNotFoundError as err:
+        assert err
+    except PDSError as err:
+        assert not err
+    else:
+        assert not "invalid codepath"
+
+    context = await build_context("local")
+
+    import random
+
+    dri1 = await pds_save_a(context, "124inmjfa0dioq-0dq" + str(random.random()))
+    dri2 = await pds_save_a(context, "1asdad24inmjfa0dioq-0dq" + str(random.random()))
+    await pds_link_dri(context, dri1, dri2)
+
+    try:
+        await pds_link_dri(context, "1wd1fc", "difhd908f")
+    except PDSRecordNotFoundError as err:
+        assert err
+    except PDSError as err:
+        assert not err
+
+
 async def pds_load_string(context, id: str, *, with_meta: bool = False) -> str:
     assert_type(id, str)
 
@@ -198,3 +255,6 @@ async def pds_oca_data_format_serialize_dict_recursive(context, dct):
     for k, v in dct.items():
         new_dict[k] = await pds_oca_data_format_serialize_item_recursive(context, k, v)
     return new_dict
+
+
+run_standalone_async(__name__, __test_pds_link)

aries_cloudagent/pdstorage_thcf/own_your_data.py

@@ -151,9 +151,6 @@ async def update_token_when_expired(self):
             await self.update_token()
 
     async def load(self, dri: str) -> dict:
-        """
-        TODO: Errors checking
-        """
         assert_type(dri, str)
         await self.update_token_when_expired()
 
@@ -167,6 +164,18 @@ async def load(self, dri: str) -> dict:
 
         return result_dict
 
+    async def link(self, source_dri, with_targets):
+        await self.update_token_when_expired()
+        body = {"source": source_dri, "targets": with_targets}
+        async with ClientSession() as session:
+            url = f"{self.api_url}/api/relation?p=dri"
+            response = await session.post(
+                url,
+                headers={"Authorization": "Bearer " + self.token["access_token"]},
+                json=body,
+            )
+            await unpack_response(response)
+
     async def save(self, record, metadata: dict, *, addition_meta={}) -> str:
         """
         meta: {
@@ -195,27 +204,20 @@ async def save(self, record, metadata: dict, *, addition_meta={}) -> str:
             "timestamp": int(round(time.time() * 1000)),  # current time in milliseconds
         }
         LOGGER.debug("OYD save record %s metadata %s", record, meta)
-        async with ClientSession() as session:
-            """
-            Pack request body
-            """
 
-            if meta.get("table") is not None:
-                table = f"{table}.{meta.get('table')}"
+        if meta.get("table") is not None:
+            table = f"{table}.{meta.get('table')}"
 
-            body = {
-                "content": record,
-                "dri": dri_value,
-                "table_name": table,
-                "mime_type": "application/json",
-            }
-            if addition_meta:
-                body.update(addition_meta)
-
-            """
-            Request
-            """
+        body = {
+            "content": record,
+            "dri": dri_value,
+            "table_name": table,
+            "mime_type": "application/json",
+        }
+        if addition_meta:
+            body.update(addition_meta)
 
+        async with ClientSession() as session:
             url = f"{self.api_url}/api/data"
             response = await session.post(
                 url,
@@ -497,15 +499,26 @@ def test_format_2():
     }, result
 
 
+async def test_link(vault):
+    await vault.link(
+        "zQmVNgyT8sYwuLby5AMp5NuPnFjwHsLVYsv2QxyUSGfrAs7",
+        ["zQmZZWT86LeNTTemdLc5vs4TybcihrHSpjxJDyCJXY1A4Lw"],
+    )
+
+
 async def test_usage_policy_parse():
     vault = OwnYourDataVault()
 
     vault.settings["client_id"] = "-s2bdkM_cv7KYDF5xg_Lj6vil1ZJaLQJ79duOW7J9g4"
     vault.settings["client_secret"] = "s_dR8dzbVES_vvc1-nyb1O_cuzyCz2_bRd3Lr12s4ug"
     vault.settings["api_url"] = "https://data-vault.eu"
+    await test_link(vault)
     await vault.update_token()
+    # result = await vault.load_multiple()
+    # print(result)
+
     # test_format()
     # test_format_2()
 
 
-run_standalone_async(__name__, test_usage_policy_parse)
+run_standalone_async(__name__, test_usage_policy_parse)

aries_cloudagent/protocols/issue_credential/v1_1/routes.py

@@ -59,20 +59,19 @@ async def issue_credential(request: web.BaseRequest):
 
     connection = await retrieve_connection(context, exchange.connection_id)
     request = exchange.credential_request
-    
+
     try:
         issuer: BaseIssuer = await context.inject(BaseIssuer)
         credential = await issuer.create_credential_ex(
             request.get("credential_values"),
             request.get("credential_type"),
-            exchange.their_public_did
+            exchange.their_public_did,
         )
     except IssuerError as err:
         raise web.HTTPInternalServerError(
             reason=f"Error occured while creating a credential {err.roll_up}"
         )
 
-
     LOGGER.info("CREDENTIAL %s", credential)
     issue = CredentialIssue(credential=credential)
     issue.assign_thread_id(exchange.thread_id)