Source code for aries_cloudagent.ledger.base

"""Ledger base class."""

from abc import ABC, abstractmethod, ABCMeta
import re
from typing import Tuple, Sequence

from ..issuer.base import BaseIssuer


[docs]class BaseLedger(ABC, metaclass=ABCMeta): """Base class for ledger.""" LEDGER_TYPE = None async def __aenter__(self) -> "BaseLedger": """ Context manager entry. Returns: The current instance """ return self async def __aexit__(self, exc_type, exc, tb): """Context manager exit."""
[docs] @abstractmethod async def get_key_for_did(self, did: str) -> str: """Fetch the verkey for a ledger DID. Args: did: The DID to look up on the ledger or in the cache """
[docs] @abstractmethod async def get_endpoint_for_did(self, did: str) -> str: """Fetch the endpoint for a ledger DID. Args: did: The DID to look up on the ledger or in the cache """
[docs] @abstractmethod async def update_endpoint_for_did(self, did: str, endpoint: str) -> bool: """Check and update the endpoint on the ledger. Args: did: The ledger DID endpoint: The endpoint address """
[docs] @abstractmethod async def register_nym( self, did: str, verkey: str, alias: str = None, role: str = None ): """ Register a nym on the ledger. Args: did: DID to register on the ledger. verkey: The verification key of the keypair. alias: Human-friendly alias to assign to the DID. role: For permissioned ledgers, what role should the new DID have. """
[docs] @abstractmethod def nym_to_did(self, nym: str) -> str: """Format a nym with the ledger's DID prefix."""
[docs] def did_to_nym(self, did: str) -> str: """Remove the ledger's DID prefix to produce a nym.""" if did: return re.sub(r"^did:\w+:", "", did)
[docs] async def get_txn_author_agreement(self, reload: bool = False): """Get the current transaction author agreement, fetching it if necessary."""
[docs] async def fetch_txn_author_agreement(self): """Fetch the current AML and TAA from the ledger."""
[docs] async def accept_txn_author_agreement( self, taa_record: dict, mechanism: str, accept_time: int = None ): """Save a new record recording the acceptance of the TAA."""
[docs] async def get_latest_txn_author_acceptance(self): """Look up the latest TAA acceptance."""
[docs] def taa_digest(self, version: str, text: str): """Generate the digest of a TAA record."""
[docs] @abstractmethod async def create_and_send_schema( self, issuer: BaseIssuer, schema_name: str, schema_version: str, attribute_names: Sequence[str], ) -> Tuple[str, dict]: """ Send schema to ledger. Args: issuer: The issuer instance to use for schema creation schema_name: The schema name schema_version: The schema version attribute_names: A list of schema attributes """
[docs] @abstractmethod async def get_revoc_reg_def(self, revoc_reg_id: str) -> dict: """Look up a revocation registry definition by ID."""
[docs] @abstractmethod async def send_revoc_reg_def(self, revoc_reg_def: dict, issuer_did: str = None): """Publish a revocation registry definition to the ledger."""
[docs] @abstractmethod async def send_revoc_reg_entry( self, revoc_reg_id: str, revoc_def_type: str, revoc_reg_entry: dict, issuer_did: str = None, ): """Publish a revocation registry entry to the ledger."""
[docs] @abstractmethod async def create_and_send_credential_definition( self, issuer: BaseIssuer, schema_id: str, signature_type: str = None, tag: str = None, support_revocation: bool = False, ) -> Tuple[str, dict]: """ Send credential definition to ledger and store relevant key matter in wallet. Args: issuer: The issuer instance to use for credential definition creation schema_id: The schema id of the schema to create cred def for signature_type: The signature type to use on the credential definition tag: Optional tag to distinguish multiple credential definitions support_revocation: Optional flag to enable revocation for this cred def """
[docs] @abstractmethod async def get_credential_definition(self, credential_definition_id: str) -> dict: """ Get a credential definition from the cache if available, otherwise the ledger. Args: credential_definition_id: The schema id of the schema to fetch cred def for """
[docs] @abstractmethod async def get_revoc_reg_delta( self, revoc_reg_id: str, timestamp_from=0, timestamp_to=None ) -> (dict, int): """Look up a revocation registry delta by ID."""
[docs] @abstractmethod async def get_schema(self, schema_id: str) -> dict: """ Get a schema from the cache if available, otherwise fetch from the ledger. Args: schema_id: The schema id (or stringified sequence number) to retrieve """
[docs] @abstractmethod async def get_revoc_reg_entry(self, revoc_reg_id: str, timestamp: int): """Get revocation registry entry by revocation registry ID and timestamp."""