Change schema info from dict to class.

Signed-off-by: jamshale <[email protected]>

acapy_agent/anoncreds/base.py

@@ -16,6 +16,7 @@
     RevRegDefResult,
 )
 from .models.schema import AnonCredsSchema, GetSchemaResult, SchemaResult
+from .models.schema_info import AnoncredsSchemaInfo
 
 T = TypeVar("T")
 
@@ -131,7 +132,7 @@ async def get_revocation_list(
         """Get a revocation list from the registry."""
 
     @abstractmethod
-    async def get_schema_info_by_id(self, schema_id: str) -> dict:
+    async def get_schema_info_by_id(self, schema_id: str) -> AnoncredsSchemaInfo:
         """Get a schema info from the registry."""
 
 

acapy_agent/anoncreds/default/did_indy/registry.py

@@ -17,6 +17,7 @@
     RevRegDefResult,
 )
 from ...models.schema import AnonCredsSchema, GetSchemaResult, SchemaResult
+from ...models.schema_info import AnoncredsSchemaInfo
 
 LOGGER = logging.getLogger(__name__)
 
@@ -119,6 +120,6 @@ async def update_revocation_list(
         """Update a revocation list on the registry."""
         raise NotImplementedError()
 
-    async def get_schema_info_by_id(self, schema_id: str) -> dict:
+    async def get_schema_info_by_id(self, schema_id: str) -> AnoncredsSchemaInfo:
         """Get a schema info from the registry."""
         return await super().get_schema_info_by_id(schema_id)

acapy_agent/anoncreds/default/did_web/registry.py

@@ -16,6 +16,7 @@
     RevRegDefResult,
 )
 from ...models.schema import AnonCredsSchema, GetSchemaResult, SchemaResult
+from ...models.schema_info import AnoncredsSchemaInfo
 
 
 class DIDWebRegistry(BaseAnonCredsResolver, BaseAnonCredsRegistrar):
@@ -114,6 +115,6 @@ async def update_revocation_list(
         """Update a revocation list on the registry."""
         raise NotImplementedError()
 
-    async def get_schema_info_by_id(self, schema_id: str) -> dict:
+    async def get_schema_info_by_id(self, schema_id: str) -> AnoncredsSchemaInfo:
         """Get a schema info from the registry."""
         return await super().get_schema_info_by_id(schema_id)

acapy_agent/anoncreds/default/legacy_indy/registry.py

@@ -80,6 +80,7 @@
     SchemaResult,
     SchemaState,
 )
+from ...models.schema_info import AnoncredsSchemaInfo
 from ...revocation import (
     CATEGORY_REV_LIST,
     CATEGORY_REV_REG_DEF,
@@ -1230,11 +1231,11 @@ async def txn_submit(
         except LedgerError as err:
             raise AnonCredsRegistrationError(err.roll_up) from err
 
-    async def get_schema_info_by_id(self, schema_id: str) -> dict:
+    async def get_schema_info_by_id(self, schema_id: str) -> AnoncredsSchemaInfo:
         """Get schema info by schema id."""
         schema_id_parts = re.match(r"^(\w+):2:([^:]+):([^:]+)$", schema_id)
-        return {
-            "issuer_id": schema_id_parts.group(1),
-            "name": schema_id_parts.group(2),
-            "version": schema_id_parts.group(3),
-        }
+        return AnoncredsSchemaInfo(
+            issuer_id=schema_id_parts.group(1),
+            name=schema_id_parts.group(2),
+            version=schema_id_parts.group(3),
+        )

acapy_agent/anoncreds/holder.py

@@ -236,9 +236,9 @@ async def _finish_store_credential(
         credential_id = credential_id or str(uuid4())
         tags = {
             "schema_id": credential_data["schema_id"],
-            "schema_issuer_id": schema_info["issuer_id"],
-            "schema_name": schema_info["name"],
-            "schema_version": schema_info["version"],
+            "schema_issuer_id": schema_info.issuer_id,
+            "schema_name": schema_info.name,
+            "schema_version": schema_info.version,
             "issuer_id": credential_definition["issuerId"],
             "cred_def_id": cred_recvd.cred_def_id,
             "rev_reg_id": cred_recvd.rev_reg_id or "None",

acapy_agent/anoncreds/models/schema_info.py

@@ -0,0 +1,26 @@
+"""This class represents schema information for anoncreds."""
+
+from typing import Optional
+
+
+class AnoncredsSchemaInfo:
+    """Represents the schema information for anonymous credentials.
+
+    Attributes:
+        issuer_id (str): The identifier of the issuer.
+        name (Optional[str]): The name of the schema. Defaults to None.
+        version (Optional[str]): The version of the schema. Defaults to None.
+
+    Args:
+        issuer_id (str): The identifier of the issuer.
+        name (Optional[str], optional): The name of the schema. Defaults to None.
+        version (Optional[str], optional): The version of the schema. Defaults to None.
+    """
+
+    def __init__(
+        self, issuer_id: str, name: Optional[str] = None, version: Optional[str] = None
+    ):
+        """Initialize the schema information."""
+        self.issuer_id = issuer_id
+        self.name = name
+        self.version = version

acapy_agent/anoncreds/registry.py

@@ -21,6 +21,7 @@
     RevRegDefResult,
 )
 from .models.schema import AnonCredsSchema, GetSchemaResult, SchemaResult
+from .models.schema_info import AnoncredsSchemaInfo
 
 LOGGER = logging.getLogger(__name__)
 
@@ -99,7 +100,7 @@ async def get_credential_definition(
             credential_definition_id,
         )
 
-    async def get_schema_info_by_id(self, schema_id: str) -> dict:
+    async def get_schema_info_by_id(self, schema_id: str) -> AnoncredsSchemaInfo:
         """Get a schema info from the registry."""
         resolver = await self._resolver_for_identifier(schema_id)
         return await resolver.get_schema_info_by_id(schema_id)