Source code for

"""Basic in-memory storage implementation of VC holder interface."""

from dateutil.parser import parse as dateutil_parser
from dateutil.parser import ParserError
from typing import Mapping, Sequence

from ...core.in_memory import InMemoryProfile

from ..in_memory import InMemoryStorage, InMemoryStorageSearch

from .base import VCHolder, VCRecordSearch
from .vc_record import VCRecord
from .xform import storage_to_vc_record, vc_to_storage_record, VC_CRED_RECORD_TYPE

[docs]class InMemoryVCHolder(VCHolder): """Basic in-memory storage class.""" def __init__(self, profile: InMemoryProfile): """Initialize the in-memory VC holder instance.""" self._profile = profile self._store = InMemoryStorage(profile)
[docs] def build_type_or_schema_query(self, uri_list: Sequence[str]) -> dict: """Build and return in-memory backend specific type_or_schema_query.""" type_or_schema_query = {"$and": []} for uri in uri_list: tag_or_list = [] tag_or_list.append({f"type:{uri}": "1"}) tag_or_list.append({f"schm:{uri}": "1"}) type_or_schema_query["$and"].append({"$or": tag_or_list}) return type_or_schema_query
[docs] async def store_credential(self, cred: VCRecord): """Add a new VC record to the store. Args: cred: The VCRecord instance to store Raises: StorageDuplicateError: If the record_id is not unique """ record = vc_to_storage_record(cred) await self._store.add_record(record)
[docs] async def retrieve_credential_by_id(self, record_id: str) -> VCRecord: """Fetch a VC record by its record ID. Raises: StorageNotFoundError: If the record is not found """ record = await self._store.get_record(VC_CRED_RECORD_TYPE, record_id) return storage_to_vc_record(record)
[docs] async def retrieve_credential_by_given_id(self, given_id: str) -> VCRecord: """Fetch a VC record by its given ID ('id' property). Raises: StorageNotFoundError: If the record is not found """ record = await self._store.find_record( VC_CRED_RECORD_TYPE, {"given_id": given_id} ) return storage_to_vc_record(record)
[docs] async def delete_credential(self, cred: VCRecord): """Remove a previously-stored VC record. Raises: StorageNotFoundError: If the record is not found """ await self._store.delete_record(vc_to_storage_record(cred))
[docs] def search_credentials( self, contexts: Sequence[str] = None, types: Sequence[str] = None, schema_ids: str = None, issuer_id: str = None, subject_ids: str = None, proof_types: Sequence[str] = None, given_id: str = None, tag_query: Mapping = None, pd_uri_list: Sequence[str] = None, ) -> "VCRecordSearch": """Start a new VC record search. Args: contexts: An inclusive list of JSON-LD contexts to match types: An inclusive list of JSON-LD types to match schema_ids: An inclusive list of credential schema identifiers issuer_id: The ID of the credential issuer subject_ids: The IDs of credential subjects all of which to match proof_types: The signature suite types used for the proof objects. given_id: The given id of the credential tag_query: A tag filter clause """ query = {} if contexts: for ctx_val in contexts: query[f"ctxt:{ctx_val}"] = "1" if types: for type_val in types: query[f"type:{type_val}"] = "1" if schema_ids: for schema_val in schema_ids: query[f"schm:{schema_val}"] = "1" if subject_ids: for subject_id in subject_ids: query[f"subj:{subject_id}"] = "1" if proof_types: for proof_type in proof_types: query[f"ptyp:{proof_type}"] = "1" if issuer_id: query["issuer_id"] = issuer_id if given_id: query["given_id"] = given_id if tag_query: query.update(tag_query) if pd_uri_list: query.update(self.build_type_or_schema_query(pd_uri_list)) search = self._store.search_records(VC_CRED_RECORD_TYPE, query) return InMemoryVCRecordSearch(search)
[docs]class InMemoryVCRecordSearch(VCRecordSearch): """In-memory search for VC records.""" def __init__(self, search: InMemoryStorageSearch): """Initialize the in-memory VC record search.""" self._search = search
[docs] async def fetch(self, max_count: int = None) -> Sequence[VCRecord]: """Fetch the next list of VC records from the store. Args: max_count: Max number of records to return. If not provided, defaults to the backend's preferred page size Returns: A list of `VCRecord` instances """ rows = await self._search.fetch(max_count) records = [storage_to_vc_record(r) for r in rows] try: records.sort( key=lambda v: dateutil_parser(v.cred_value.get("issuanceDate")), reverse=True, ) return records except ParserError: return records