Module `mydata_did.patched_protocols.issue_credential.v1_0.models.credential_exchange`

Aries#0036 v1.0 credential exchange information with non-secrets storage.

Expand source code

"""Aries#0036 v1.0 credential exchange information with non-secrets storage."""

from os import environ
from typing import Any

from marshmallow import fields, validate

from aries_cloudagent.config.injection_context import InjectionContext
from aries_cloudagent.messaging.models.base_record import BaseExchangeRecord, BaseExchangeSchema
from aries_cloudagent.messaging.valid import INDY_CRED_DEF_ID, INDY_SCHEMA_ID, UUIDFour

unencrypted_tags = environ.get("EXCH_UNENCRYPTED_TAGS", "False").upper() == "TRUE"


class V10CredentialExchange(BaseExchangeRecord):
    """Represents an Aries#0036 credential exchange."""

    class Meta:
        """CredentialExchange metadata."""

        schema_class = "V10CredentialExchangeSchema"

    RECORD_TYPE = "credential_exchange_v10"
    RECORD_ID_NAME = "credential_exchange_id"
    WEBHOOK_TOPIC = "issue_credential"
    TAG_NAMES = {"~thread_id"} if unencrypted_tags else {"thread_id"}

    INITIATOR_SELF = "self"
    INITIATOR_EXTERNAL = "external"
    ROLE_ISSUER = "issuer"
    ROLE_HOLDER = "holder"

    STATE_PROPOSAL_SENT = "proposal_sent"
    STATE_PROPOSAL_RECEIVED = "proposal_received"
    STATE_OFFER_SENT = "offer_sent"
    STATE_OFFER_RECEIVED = "offer_received"
    STATE_REQUEST_SENT = "request_sent"
    STATE_REQUEST_RECEIVED = "request_received"
    STATE_ISSUED = "credential_issued"
    STATE_CREDENTIAL_RECEIVED = "credential_received"
    STATE_ACKED = "credential_acked"

    def __init__(
        self,
        *,
        credential_exchange_id: str = None,
        connection_id: str = None,
        thread_id: str = None,
        parent_thread_id: str = None,
        initiator: str = None,
        role: str = None,
        state: str = None,
        credential_definition_id: str = None,
        schema_id: str = None,
        credential_proposal_dict: dict = None,  # serialized credential proposal message
        credential_offer_dict: dict = None,  # serialized credential offer message
        credential_offer: dict = None,  # indy credential offer
        credential_request: dict = None,  # indy credential request
        credential_request_metadata: dict = None,
        credential_id: str = None,
        raw_credential: dict = None,  # indy credential as received
        credential: dict = None,  # indy credential as stored
        revoc_reg_id: str = None,
        revocation_id: str = None,
        auto_offer: bool = False,
        auto_issue: bool = False,
        auto_remove: bool = True,
        error_msg: str = None,
        trace: bool = False,
        template_id: str = None,
        **kwargs,
    ):
        """Initialize a new V10CredentialExchange."""
        super().__init__(credential_exchange_id, state, trace=trace, **kwargs)
        self._id = credential_exchange_id
        self.connection_id = connection_id
        self.thread_id = thread_id
        self.parent_thread_id = parent_thread_id
        self.initiator = initiator
        self.role = role
        self.state = state
        self.credential_definition_id = credential_definition_id
        self.schema_id = schema_id
        self.credential_proposal_dict = credential_proposal_dict
        self.credential_offer_dict = credential_offer_dict
        self.credential_offer = credential_offer
        self.credential_request = credential_request
        self.credential_request_metadata = credential_request_metadata
        self.credential_id = credential_id
        self.raw_credential = raw_credential
        self.credential = credential
        self.revoc_reg_id = revoc_reg_id
        self.revocation_id = revocation_id
        self.auto_offer = auto_offer
        self.auto_issue = auto_issue
        self.auto_remove = auto_remove
        self.error_msg = error_msg
        self.trace = trace
        self.template_id = template_id

    @property
    def credential_exchange_id(self) -> str:
        """Accessor for the ID associated with this exchange."""
        return self._id

    @property
    def record_value(self) -> dict:
        """Accessor for the JSON record value generated for this credential exchange."""
        return {
            prop: getattr(self, prop)
            for prop in (
                "connection_id",
                "credential_proposal_dict",
                "credential_offer_dict",
                "credential_offer",
                "credential_request",
                "credential_request_metadata",
                "error_msg",
                "auto_offer",
                "auto_issue",
                "auto_remove",
                "raw_credential",
                "credential",
                "parent_thread_id",
                "initiator",
                "credential_definition_id",
                "schema_id",
                "credential_id",
                "revoc_reg_id",
                "revocation_id",
                "role",
                "state",
                "trace",
                "template_id"
            )
        }

    @classmethod
    async def retrieve_by_connection_and_thread(
        cls, context: InjectionContext, connection_id: str, thread_id: str
    ) -> "V10CredentialExchange":
        """Retrieve a credential exchange record by connection and thread ID."""
        cache_key = f"credential_exchange_ctidx::{connection_id}::{thread_id}"
        record_id = await cls.get_cached_key(context, cache_key)
        if record_id:
            record = await cls.retrieve_by_id(context, record_id)
        else:
            record = await cls.retrieve_by_tag_filter(
                context,
                {"thread_id": thread_id},
                {"connection_id": connection_id} if connection_id else None,
            )
            await cls.set_cached_key(context, cache_key, record.credential_exchange_id)
        return record

    def __eq__(self, other: Any) -> bool:
        """Comparison between records."""
        return super().__eq__(other)


class V10CredentialExchangeSchema(BaseExchangeSchema):
    """Schema to allow serialization/deserialization of credential exchange records."""

    class Meta:
        """V10CredentialExchangeSchema metadata."""

        model_class = V10CredentialExchange

    credential_exchange_id = fields.Str(
        required=False,
        description="Credential exchange identifier",
        example=UUIDFour.EXAMPLE,
    )
    connection_id = fields.Str(
        required=False, description="Connection identifier", example=UUIDFour.EXAMPLE
    )
    thread_id = fields.Str(
        required=False, description="Thread identifier", example=UUIDFour.EXAMPLE
    )
    parent_thread_id = fields.Str(
        required=False, description="Parent thread identifier", example=UUIDFour.EXAMPLE
    )
    initiator = fields.Str(
        required=False,
        description="Issue-credential exchange initiator: self or external",
        example=V10CredentialExchange.INITIATOR_SELF,
        validate=validate.OneOf(["self", "external"]),
    )
    role = fields.Str(
        required=False,
        description="Issue-credential exchange role: holder or issuer",
        example=V10CredentialExchange.ROLE_ISSUER,
        validate=validate.OneOf(["holder", "issuer"]),
    )
    state = fields.Str(
        required=False,
        description="Issue-credential exchange state",
        example=V10CredentialExchange.STATE_ACKED,
    )
    credential_definition_id = fields.Str(
        required=False,
        description="Credential definition identifier",
        **INDY_CRED_DEF_ID,
    )
    schema_id = fields.Str(
        required=False, description="Schema identifier", **INDY_SCHEMA_ID
    )
    credential_proposal_dict = fields.Dict(
        required=False, description="Serialized credential proposal message"
    )
    credential_offer_dict = fields.Dict(
        required=False, description="Serialized credential offer message"
    )
    credential_offer = fields.Dict(
        required=False, description="(Indy) credential offer"
    )
    credential_request = fields.Dict(
        required=False, description="(Indy) credential request"
    )
    credential_request_metadata = fields.Dict(
        required=False, description="(Indy) credential request metadata"
    )
    credential_id = fields.Str(
        required=False, description="Credential identifier", example=UUIDFour.EXAMPLE
    )
    raw_credential = fields.Dict(
        required=False,
        description="Credential as received, prior to storage in holder wallet",
    )
    credential = fields.Dict(required=False, description="Credential as stored")
    auto_offer = fields.Bool(
        required=False,
        description="Holder choice to accept offer in this credential exchange",
        example=False,
    )
    auto_issue = fields.Bool(
        required=False,
        description="Issuer choice to issue to request in this credential exchange",
        example=False,
    )
    auto_remove = fields.Bool(
        required=False,
        default=True,
        description=(
            "Issuer choice to remove this credential exchange record when complete"
        ),
        example=False,
    )
    error_msg = fields.Str(
        required=False,
        description="Error message",
        example="credential definition identifier is not set in proposal",
    )
    revoc_reg_id = fields.Str(
        required=False, description="Revocation registry identifier"
    )
    revocation_id = fields.Str(
        required=False, description="Credential identifier within revocation registry"
    )
    template_id = fields.Str(required=False)

Classes

class V10CredentialExchange (*, credential_exchange_id: str = None, connection_id: str = None, thread_id: str = None, parent_thread_id: str = None, initiator: str = None, role: str = None, state: str = None, credential_definition_id: str = None, schema_id: str = None, credential_proposal_dict: dict = None, credential_offer_dict: dict = None, credential_offer: dict = None, credential_request: dict = None, credential_request_metadata: dict = None, credential_id: str = None, raw_credential: dict = None, credential: dict = None, revoc_reg_id: str = None, revocation_id: str = None, auto_offer: bool = False, auto_issue: bool = False, auto_remove: bool = True, error_msg: str = None, trace: bool = False, template_id: str = None, **kwargs)

Represents an Aries#0036 credential exchange.

Initialize a new V10CredentialExchange.

Expand source code

class V10CredentialExchange(BaseExchangeRecord):
    """Represents an Aries#0036 credential exchange."""

    class Meta:
        """CredentialExchange metadata."""

        schema_class = "V10CredentialExchangeSchema"

    RECORD_TYPE = "credential_exchange_v10"
    RECORD_ID_NAME = "credential_exchange_id"
    WEBHOOK_TOPIC = "issue_credential"
    TAG_NAMES = {"~thread_id"} if unencrypted_tags else {"thread_id"}

    INITIATOR_SELF = "self"
    INITIATOR_EXTERNAL = "external"
    ROLE_ISSUER = "issuer"
    ROLE_HOLDER = "holder"

    STATE_PROPOSAL_SENT = "proposal_sent"
    STATE_PROPOSAL_RECEIVED = "proposal_received"
    STATE_OFFER_SENT = "offer_sent"
    STATE_OFFER_RECEIVED = "offer_received"
    STATE_REQUEST_SENT = "request_sent"
    STATE_REQUEST_RECEIVED = "request_received"
    STATE_ISSUED = "credential_issued"
    STATE_CREDENTIAL_RECEIVED = "credential_received"
    STATE_ACKED = "credential_acked"

    def __init__(
        self,
        *,
        credential_exchange_id: str = None,
        connection_id: str = None,
        thread_id: str = None,
        parent_thread_id: str = None,
        initiator: str = None,
        role: str = None,
        state: str = None,
        credential_definition_id: str = None,
        schema_id: str = None,
        credential_proposal_dict: dict = None,  # serialized credential proposal message
        credential_offer_dict: dict = None,  # serialized credential offer message
        credential_offer: dict = None,  # indy credential offer
        credential_request: dict = None,  # indy credential request
        credential_request_metadata: dict = None,
        credential_id: str = None,
        raw_credential: dict = None,  # indy credential as received
        credential: dict = None,  # indy credential as stored
        revoc_reg_id: str = None,
        revocation_id: str = None,
        auto_offer: bool = False,
        auto_issue: bool = False,
        auto_remove: bool = True,
        error_msg: str = None,
        trace: bool = False,
        template_id: str = None,
        **kwargs,
    ):
        """Initialize a new V10CredentialExchange."""
        super().__init__(credential_exchange_id, state, trace=trace, **kwargs)
        self._id = credential_exchange_id
        self.connection_id = connection_id
        self.thread_id = thread_id
        self.parent_thread_id = parent_thread_id
        self.initiator = initiator
        self.role = role
        self.state = state
        self.credential_definition_id = credential_definition_id
        self.schema_id = schema_id
        self.credential_proposal_dict = credential_proposal_dict
        self.credential_offer_dict = credential_offer_dict
        self.credential_offer = credential_offer
        self.credential_request = credential_request
        self.credential_request_metadata = credential_request_metadata
        self.credential_id = credential_id
        self.raw_credential = raw_credential
        self.credential = credential
        self.revoc_reg_id = revoc_reg_id
        self.revocation_id = revocation_id
        self.auto_offer = auto_offer
        self.auto_issue = auto_issue
        self.auto_remove = auto_remove
        self.error_msg = error_msg
        self.trace = trace
        self.template_id = template_id

    @property
    def credential_exchange_id(self) -> str:
        """Accessor for the ID associated with this exchange."""
        return self._id

    @property
    def record_value(self) -> dict:
        """Accessor for the JSON record value generated for this credential exchange."""
        return {
            prop: getattr(self, prop)
            for prop in (
                "connection_id",
                "credential_proposal_dict",
                "credential_offer_dict",
                "credential_offer",
                "credential_request",
                "credential_request_metadata",
                "error_msg",
                "auto_offer",
                "auto_issue",
                "auto_remove",
                "raw_credential",
                "credential",
                "parent_thread_id",
                "initiator",
                "credential_definition_id",
                "schema_id",
                "credential_id",
                "revoc_reg_id",
                "revocation_id",
                "role",
                "state",
                "trace",
                "template_id"
            )
        }

    @classmethod
    async def retrieve_by_connection_and_thread(
        cls, context: InjectionContext, connection_id: str, thread_id: str
    ) -> "V10CredentialExchange":
        """Retrieve a credential exchange record by connection and thread ID."""
        cache_key = f"credential_exchange_ctidx::{connection_id}::{thread_id}"
        record_id = await cls.get_cached_key(context, cache_key)
        if record_id:
            record = await cls.retrieve_by_id(context, record_id)
        else:
            record = await cls.retrieve_by_tag_filter(
                context,
                {"thread_id": thread_id},
                {"connection_id": connection_id} if connection_id else None,
            )
            await cls.set_cached_key(context, cache_key, record.credential_exchange_id)
        return record

    def __eq__(self, other: Any) -> bool:
        """Comparison between records."""
        return super().__eq__(other)

Ancestors

aries_cloudagent.messaging.models.base_record.BaseExchangeRecord
aries_cloudagent.messaging.models.base_record.BaseRecord
aries_cloudagent.messaging.models.base.BaseModel
abc.ABC

Class variables

var INITIATOR_EXTERNAL
var INITIATOR_SELF
var Meta: CredentialExchange metadata.
var RECORD_ID_NAME
var RECORD_TYPE
var ROLE_HOLDER
var ROLE_ISSUER
var STATE_ACKED
var STATE_CREDENTIAL_RECEIVED
var STATE_ISSUED
var STATE_OFFER_RECEIVED
var STATE_OFFER_SENT
var STATE_PROPOSAL_RECEIVED
var STATE_PROPOSAL_SENT
var STATE_REQUEST_RECEIVED
var STATE_REQUEST_SENT
var TAG_NAMES
var WEBHOOK_TOPIC

Static methods

async def retrieve_by_connection_and_thread(context: aries_cloudagent.config.injection_context.InjectionContext, connection_id: str, thread_id: str) ‑> V10CredentialExchange

Retrieve a credential exchange record by connection and thread ID.

Expand source code

@classmethod
async def retrieve_by_connection_and_thread(
    cls, context: InjectionContext, connection_id: str, thread_id: str
) -> "V10CredentialExchange":
    """Retrieve a credential exchange record by connection and thread ID."""
    cache_key = f"credential_exchange_ctidx::{connection_id}::{thread_id}"
    record_id = await cls.get_cached_key(context, cache_key)
    if record_id:
        record = await cls.retrieve_by_id(context, record_id)
    else:
        record = await cls.retrieve_by_tag_filter(
            context,
            {"thread_id": thread_id},
            {"connection_id": connection_id} if connection_id else None,
        )
        await cls.set_cached_key(context, cache_key, record.credential_exchange_id)
    return record

Instance variables

var credential_exchange_id : str

Accessor for the ID associated with this exchange.

Expand source code

@property
def credential_exchange_id(self) -> str:
    """Accessor for the ID associated with this exchange."""
    return self._id

var record_value : dict

Accessor for the JSON record value generated for this credential exchange.

Expand source code

@property
def record_value(self) -> dict:
    """Accessor for the JSON record value generated for this credential exchange."""
    return {
        prop: getattr(self, prop)
        for prop in (
            "connection_id",
            "credential_proposal_dict",
            "credential_offer_dict",
            "credential_offer",
            "credential_request",
            "credential_request_metadata",
            "error_msg",
            "auto_offer",
            "auto_issue",
            "auto_remove",
            "raw_credential",
            "credential",
            "parent_thread_id",
            "initiator",
            "credential_definition_id",
            "schema_id",
            "credential_id",
            "revoc_reg_id",
            "revocation_id",
            "role",
            "state",
            "trace",
            "template_id"
        )
    }

class V10CredentialExchangeSchema (*args, **kwargs)

Schema to allow serialization/deserialization of credential exchange records.

Initialize BaseModelSchema.

Raises

TypeError: If model_class is not set on Meta

Expand source code

class V10CredentialExchangeSchema(BaseExchangeSchema):
    """Schema to allow serialization/deserialization of credential exchange records."""

    class Meta:
        """V10CredentialExchangeSchema metadata."""

        model_class = V10CredentialExchange

    credential_exchange_id = fields.Str(
        required=False,
        description="Credential exchange identifier",
        example=UUIDFour.EXAMPLE,
    )
    connection_id = fields.Str(
        required=False, description="Connection identifier", example=UUIDFour.EXAMPLE
    )
    thread_id = fields.Str(
        required=False, description="Thread identifier", example=UUIDFour.EXAMPLE
    )
    parent_thread_id = fields.Str(
        required=False, description="Parent thread identifier", example=UUIDFour.EXAMPLE
    )
    initiator = fields.Str(
        required=False,
        description="Issue-credential exchange initiator: self or external",
        example=V10CredentialExchange.INITIATOR_SELF,
        validate=validate.OneOf(["self", "external"]),
    )
    role = fields.Str(
        required=False,
        description="Issue-credential exchange role: holder or issuer",
        example=V10CredentialExchange.ROLE_ISSUER,
        validate=validate.OneOf(["holder", "issuer"]),
    )
    state = fields.Str(
        required=False,
        description="Issue-credential exchange state",
        example=V10CredentialExchange.STATE_ACKED,
    )
    credential_definition_id = fields.Str(
        required=False,
        description="Credential definition identifier",
        **INDY_CRED_DEF_ID,
    )
    schema_id = fields.Str(
        required=False, description="Schema identifier", **INDY_SCHEMA_ID
    )
    credential_proposal_dict = fields.Dict(
        required=False, description="Serialized credential proposal message"
    )
    credential_offer_dict = fields.Dict(
        required=False, description="Serialized credential offer message"
    )
    credential_offer = fields.Dict(
        required=False, description="(Indy) credential offer"
    )
    credential_request = fields.Dict(
        required=False, description="(Indy) credential request"
    )
    credential_request_metadata = fields.Dict(
        required=False, description="(Indy) credential request metadata"
    )
    credential_id = fields.Str(
        required=False, description="Credential identifier", example=UUIDFour.EXAMPLE
    )
    raw_credential = fields.Dict(
        required=False,
        description="Credential as received, prior to storage in holder wallet",
    )
    credential = fields.Dict(required=False, description="Credential as stored")
    auto_offer = fields.Bool(
        required=False,
        description="Holder choice to accept offer in this credential exchange",
        example=False,
    )
    auto_issue = fields.Bool(
        required=False,
        description="Issuer choice to issue to request in this credential exchange",
        example=False,
    )
    auto_remove = fields.Bool(
        required=False,
        default=True,
        description=(
            "Issuer choice to remove this credential exchange record when complete"
        ),
        example=False,
    )
    error_msg = fields.Str(
        required=False,
        description="Error message",
        example="credential definition identifier is not set in proposal",
    )
    revoc_reg_id = fields.Str(
        required=False, description="Revocation registry identifier"
    )
    revocation_id = fields.Str(
        required=False, description="Credential identifier within revocation registry"
    )
    template_id = fields.Str(required=False)

Ancestors

aries_cloudagent.messaging.models.base_record.BaseExchangeSchema
aries_cloudagent.messaging.models.base_record.BaseRecordSchema
aries_cloudagent.messaging.models.base.BaseModelSchema
marshmallow.schema.Schema
marshmallow.base.SchemaABC

Class variables

var Meta: V10CredentialExchangeSchema metadata.
var opts