Add ping method to PDS API, Add a user credentials validity check on …

…settings stage

aries_cloudagent/holder/pds.py

@@ -16,7 +16,7 @@
 )
 from .base import BaseHolder, HolderError
 from aries_cloudagent.pdstorage_thcf.api import load_string, load_table, save_string
-from aries_cloudagent.pdstorage_thcf.error import PersonalDataStorageNotFoundError
+from aries_cloudagent.pdstorage_thcf.error import PDSNotFoundError
 
 CREDENTIALS_TABLE = "credentials"
 
@@ -38,7 +38,7 @@ async def get_credential(self, credential_id: str) -> str:
         """
         try:
             credential = await load_string(self.context, credential_id)
-        except PersonalDataStorageNotFoundError as err:
+        except PDSNotFoundError as err:
             raise HolderError(err.roll_up)
 
         return credential
@@ -249,7 +249,7 @@ async def store_credential(
                 json.dumps(credential_data),
                 metadata=json.dumps({"table": CREDENTIALS_TABLE}),
             )
-        except PersonalDataStorageNotFoundError as err:
+        except PDSNotFoundError as err:
             raise HolderError(err.roll_up)
         return record_id
 
@@ -263,7 +263,7 @@ async def get_credentials(self) -> list:
         """
         try:
             query = await load_table(self.context, CREDENTIALS_TABLE)
-        except PersonalDataStorageNotFoundError as err:
+        except PDSNotFoundError as err:
             raise HolderError(err.roll_up)
 
         query = json.loads(query)

aries_cloudagent/pdstorage_thcf/api.py

@@ -1,5 +1,5 @@
 from .base import BasePersonalDataStorage
-from .error import PersonalDataStorageNotFoundError
+from .error import PDSNotFoundError
 from .models.saved_personal_storage import SavedPersonalStorage
 import hashlib
 import multihash
@@ -16,7 +16,7 @@ async def pds_get_active_pds_name(context):
     try:
         active_pds = await SavedPersonalStorage.retrieve_active(context)
     except StorageNotFoundError as err:
-        raise PersonalDataStorageNotFoundError(f"No active pds found {err.roll_up}")
+        raise PDSNotFoundError(f"No active pds found {err.roll_up}")
 
     return active_pds.get_pds_name()
 
@@ -35,7 +35,7 @@ async def load_string(context, id: str) -> str:
         )
         debug_all_records = await DriStorageMatchTable.query(context)
         LOGGER.error("All records in table: ", debug_all_records)
-        raise PersonalDataStorageNotFoundError(err)
+        raise PDSNotFoundError(err)
 
     pds: BasePersonalDataStorage = await context.inject(
         BasePersonalDataStorage, {"personal_storage_type": match.pds_type}
@@ -89,7 +89,7 @@ async def delete_record(context, id: str) -> str:
             f"plugin: {match}",
             f"ERROR: {err.roll_up}",
         )
-        raise PersonalDataStorageNotFoundError(err)
+        raise PDSNotFoundError(err)
 
     pds: BasePersonalDataStorage = await context.inject(
         BasePersonalDataStorage, {"personal_storage_type": match.pds_type}

aries_cloudagent/pdstorage_thcf/base.py

@@ -8,20 +8,24 @@ def __init__(self):
 
     @abstractmethod
     async def save(self, record: str, metadata: str) -> str:
-        """
-        Returns: saved data id, (should maybe return None on key not found?)
-
-        """
+        """Returns: saved data id, (should maybe return None on key not found?)."""
 
     @abstractmethod
     async def load(self, id: str) -> str:
-        """
-        Returns: data represented by id
-        """
+        """Returns: data represented by id."""
 
     @abstractmethod
     async def load_table(self, table: str) -> str:
-        """"""
+        """Load all records from a table."""
+
+    @abstractmethod
+    async def ping(self) -> [bool, str]:
+        """
+        Returns: true if we connected at all, false if service is not responding.
+                 and additional info about the failure
+
+        connected, exception = await personal_storage.ping()
+        """
 
     def __repr__(self) -> str:
         """

aries_cloudagent/pdstorage_thcf/data_vault.py

@@ -1,6 +1,6 @@
 from .base import BasePersonalDataStorage
-from .error import PersonalDataStorageNotFoundError
-from aiohttp import ClientSession, FormData
+from .error import PDSNotFoundError
+from aiohttp import ClientSession, FormData, ClientConnectionError
 import json
 import logging
 
@@ -53,7 +53,7 @@ async def save(self, record: str, metadata: str) -> str:
         data.add_field("file", record, filename="data", content_type="application/json")
         url = f"{self.settings['api_url']}{API_ENDPOINT}"
         LOGGER.info(
-            f"""DataVault.save: 
+            f"""DataVault.save:
                 url: {url}
                 id: {id}
                 settings: {self.settings}
@@ -69,4 +69,16 @@ async def save(self, record: str, metadata: str) -> str:
 
     async def load_table(self, table: str) -> str:
 
-        return "tables not supported by active PDStorage "
+        return "tables not supported by active PDStorage"
+
+    async def ping(self) -> [bool, str]:
+        """
+        Returns: true if we connected at all, false if service is not responding.
+                 and additional info about the failure
+        """
+        try:
+            await self.load("ping")
+        except ClientConnectionError as err:
+            return [False, str(err)]
+
+        return [True, None]

aries_cloudagent/pdstorage_thcf/error.py

@@ -1,9 +1,16 @@
 from ..core.error import BaseError
 
 
-class PersonalDataStorageError(BaseError):
+class PDSError(BaseError):
     """Base class for Storage errors."""
 
 
-class PersonalDataStorageNotFoundError(BaseError):
-    """Class for record not found in storage"""
+class PDSNotFoundError(PDSError):
+    """
+    Class of PDS not found.
+    This can be thrown when there is no active PDS or PDS type was not found.
+    """
+
+
+class PDSRecordNotFoundError(PDSError):
+    """Record not found in storage"""

aries_cloudagent/pdstorage_thcf/handlers.py

@@ -1,6 +1,6 @@
 from .api import load_string, save_string
 from .base import *
-from .error import PersonalDataStorageError, PersonalDataStorageNotFoundError
+from .error import PDSError, PDSNotFoundError
 from .message_types import ExchangeDataB, ExchangeDataA
 from aries_cloudagent.messaging.base_handler import (
     BaseHandler,
@@ -25,8 +25,8 @@ async def handle(self, context: RequestContext, responder: BaseResponder):
         try:
             payload = await load_string(context, payload_dri)
             if payload == None:
-                raise PersonalDataStorageNotFoundError
-        except PersonalDataStorageNotFoundError as err:
+                raise PDSNotFoundError
+        except PDSNotFoundError as err:
             LOGGER.warning("TODO: ExchangeDataAHandler ProblemReport %s", err.roll_up)
             return
 
@@ -53,7 +53,7 @@ async def handle(self, context: RequestContext, responder: BaseResponder):
         """
         try:
             payload_dri = await save_string(context, context.message.payload)
-        except PersonalDataStorageError as err:
+        except PDSError as err:
             raise err.roll_up
 
         if context.message.payload_dri:

aries_cloudagent/pdstorage_thcf/local.py

@@ -1,5 +1,5 @@
 from .base import BasePersonalDataStorage
-from .error import PersonalDataStorageNotFoundError
+from .error import PDSNotFoundError
 from .api import encode
 
 
@@ -31,3 +31,6 @@ async def save(self, record: str, metadata: str) -> str:
     async def load_table(self, table: str) -> str:
 
         return "tables not supported by active PDStorage "
+
+    async def ping(self) -> [bool, str]:
+        return [True, None]

aries_cloudagent/pdstorage_thcf/own_your_data.py

@@ -1,12 +1,12 @@
 from .base import BasePersonalDataStorage
 from .api import encode
-from .error import PersonalDataStorageError
+from .error import PDSError, PDSRecordNotFoundError
 
 import json
 import logging
 from urllib.parse import urlparse
 
-from aiohttp import ClientSession
+from aiohttp import ClientSession, ClientConnectionError
 from aries_cloudagent.aathcf.credentials import assert_type
 import time
 from collections import OrderedDict
@@ -16,9 +16,13 @@
 
 async def unpack_response(response):
     result: str = await response.text()
-    if response.status != 200:
+    if response.status == 404:
         LOGGER.error("Error Own Your Data PDS", result)
-        raise PersonalDataStorageError("Error Own Your Data PDS", result)
+        raise PDSRecordNotFoundError("Record not found in Own your data PDS", result)
+
+    elif response.status != 200:
+        LOGGER.error("Error Own Your Data PDS", result)
+        raise PDSError("Error Own Your Data PDS", result)
 
     return result
 
@@ -41,75 +45,66 @@ async def get_usage_policy(self):
         return self.settings["usage_policy"]
 
     async def update_token(self):
+        parsed_url = urlparse(self.settings.get("api_url"))
+        self.api_url = "{url.scheme}://{url.netloc}".format(url=parsed_url)
+        LOGGER.debug("API URL OYD %s", self.api_url)
+
+        client_id = self.settings.get("client_id")
+        client_secret = self.settings.get("client_secret")
+        grant_type = self.settings.get("grant_type", "client_credentials")
+        scope = self.settings.get("scope")
+
+        if self.api_url is None:
+            raise PDSError("Please configure the plugin, api_url is empty")
+        if client_id is None:
+            raise PDSError("Please configure the plugin, client_id is empty")
+        if client_secret is None:
+            raise PDSError("Please configure the plugin, client_secret is empty")
+
+        async with ClientSession() as session:
+            body = {
+                "client_id": client_id,
+                "client_secret": client_secret,
+                "grant_type": grant_type,
+            }
+            if scope is not None:
+                body["scope"] = scope
+            result = await session.post(
+                self.api_url + "/oauth/token",
+                json=body,
+            )
+            result = await unpack_response(result)
+            token = json.loads(result)
+            self.token = token
+            self.token_timestamp = time.time()
+            LOGGER.info("update token: %s", self.token)
 
         """
-        Check if the token expired
+        Download the usage policy
+
         """
 
+        url = f"{self.api_url}/api/meta/usage"
+        async with ClientSession() as session:
+            result = await session.get(
+                url,
+                headers={"Authorization": "Bearer " + self.token["access_token"]},
+            )
+            result = await unpack_response(result)
+            self.settings["usage_policy"] = result
+            LOGGER.debug("Usage policy %s", self.settings["usage_policy"])
+
+    async def update_token_when_expired(self):
         time_elapsed = time.time() - (self.token_timestamp - 10)
         if time_elapsed > float(self.token["expires_in"]):
-
-            parsed_url = urlparse(self.settings.get("api_url"))
-            self.api_url = "{url.scheme}://{url.netloc}".format(url=parsed_url)
-            LOGGER.debug("API URL OYD %s", self.api_url)
-
-            client_id = self.settings.get("client_id")
-            client_secret = self.settings.get("client_secret")
-            grant_type = self.settings.get("grant_type", "client_credentials")
-            scope = self.settings.get("scope")
-
-            if self.api_url is None:
-                raise PersonalDataStorageError(
-                    "Please configure the plugin, api_url is empty"
-                )
-            if client_id is None:
-                raise PersonalDataStorageError(
-                    "Please configure the plugin, client_id is empty"
-                )
-            if client_secret is None:
-                raise PersonalDataStorageError(
-                    "Please configure the plugin, client_secret is empty"
-                )
-
-            async with ClientSession() as session:
-                body = {
-                    "client_id": client_id,
-                    "client_secret": client_secret,
-                    "grant_type": grant_type,
-                }
-                if scope is not None:
-                    body["scope"] = scope
-                result = await session.post(
-                    self.api_url + "/oauth/token",
-                    json=body,
-                )
-                result = await unpack_response(result)
-                token = json.loads(result)
-                self.token = token
-                self.token_timestamp = time.time()
-                LOGGER.debug("update token: %s", self.token)
-
-            """
-            Download the usage policy
-
-            """
-
-            url = f"{self.api_url}/api/meta/usage"
-            async with ClientSession() as session:
-                result = await session.get(
-                    url,
-                    headers={"Authorization": "Bearer " + self.token["access_token"]},
-                )
-                result = await unpack_response(result)
-                self.settings["usage_policy"] = result
-                LOGGER.debug("Usage policy %s", self.settings["usage_policy"])
+            await self.update_token()
 
     async def load(self, dri: str) -> str:
         """
         TODO: Errors checking
         """
         assert_type(dri, str)
-        await self.update_token()
+        await self.update_token_when_expired()
 
         url = f"{self.api_url}/api/data/{dri}?p=dri&f=plain"
         async with ClientSession() as session:
@@ -134,7 +129,7 @@ async def save(self, record: str, metadata: str) -> str:
         """
         assert_type(record, str)
         assert_type(metadata, str)
-        await self.update_token()
+        await self.update_token_when_expired()
 
         table = self.settings.get("repo")
         table = table if table is not None else "dip.data"
@@ -183,7 +178,7 @@ async def save(self, record: str, metadata: str) -> str:
 
     async def load_table(self, table: str) -> str:
         assert_type(table, str)
-        await self.update_token()
+        await self.update_token_when_expired()
 
         url = f"{self.api_url}/api/data?table=dip.data.{table}&f=plain"
         LOGGER.debug("OYD LOAD TABLE url [ %s ]", url)
@@ -195,3 +190,13 @@ async def load_table(self, table: str) -> str:
             LOGGER.debug("OYD LOAD TABLE result: [ %s ]", result)
 
         return result
+
+    async def ping(self) -> [bool, str]:
+        try:
+            await self.update_token()
+        except ClientConnectionError as err:
+            return [False, str(err)]
+        except PDSError as err:
+            return [False, str(err)]
+
+        return [True, None]