"""Abstract interfaces for VC holder implementations."""
from abc import ABC, abstractmethod
from typing import Mapping, Sequence
from .vc_record import VCRecord
[docs]class VCHolder(ABC):
"""Abstract base class for a verifiable credential holder."""
[docs] @abstractmethod
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
"""
[docs] @abstractmethod
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
"""
[docs] @abstractmethod
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
"""
[docs] @abstractmethod
async def delete_credential(self, cred: VCRecord):
"""Remove a previously-stored VC record.
Raises:
StorageNotFoundError: If the record is not found
"""
[docs] @abstractmethod
def build_type_or_schema_query(self, uri_list: Sequence[str]) -> dict:
"""Build and return backend-specific type_or_schema_query.
Args:
uri_list: List of schema uri from input_descriptor
"""
[docs] @abstractmethod
def search_credentials(
self,
contexts: Sequence[str] = None,
types: Sequence[str] = None,
schema_ids: Sequence[str] = None,
issuer_id: str = None,
subject_ids: Sequence[str] = None,
proof_types: Sequence[str] = None,
given_id: str = None,
tag_query: Mapping = 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 any 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
"""
def __repr__(self) -> str:
"""Return a human readable representation of this class.
Returns:
A human readable string for this class
"""
return "<{}>".format(self.__class__.__name__)
[docs]class VCRecordSearch(ABC):
"""A VC record search in progress."""
[docs] @abstractmethod
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
"""
[docs] async def close(self):
"""Dispose of the search query."""
def __aiter__(self):
"""Async iterator magic method."""
return IterVCRecordSearch(self)
def __repr__(self) -> str:
"""Human readable representation of this instance."""
return "<{}>".format(self.__class__.__name__)
[docs]class IterVCRecordSearch:
"""A generic record search async iterator."""
def __init__(self, search: VCRecordSearch, page_size: int = None):
"""Instantiate a new `IterVCRecordSearch` instance."""
self._buffer = None
self._page_size = page_size
self._search = search
async def __anext__(self):
"""Async iterator magic method."""
if not self._buffer:
self._buffer = await self._search.fetch(self._page_size) or []
try:
return self._buffer.pop(0)
except IndexError:
raise StopAsyncIteration