Module dexa_sdk.marketplace.records.marketplace_connection_record

Classes

class MarketplaceConnectionRecord (id: str = None, connection_id: str = None, state: str = None, **kwargs)

Marketplace connection record model

Initialize a new BaseRecord.

Expand source code

class MarketplaceConnectionRecord(BaseRecord):
    """Marketplace connection record model"""
    class Meta:
        schema_class = "MarketplaceConnectionRecordSchema"

    # Record type
    RECORD_TYPE = "marketplace_connection"

    # Record identifier
    RECORD_ID_NAME = "id"

    # Webhook topic name for this record type
    WEBHOOK_TOPIC = None

    # Record tags
    TAG_NAMES = {
        "~connection_id"
    }

    def __init__(
        self,
        id: str = None,
        connection_id: str = None,
        state: str = None,
        **kwargs
    ):
        # Pass the identifier and state to parent class
        super().__init__(id, state, **kwargs)

        self.connection_id = connection_id

    @property
    def record_value(self) -> dict:
        """Accessor for JSON record value generated for this transaction record."""
        return {
            prop: getattr(self, prop)
            for prop in (
                "connection_id",
                "state"
            )
        }

    async def controller_details_model(
        self,
        context: InjectionContext
    ) -> ControllerDetailModel:
        """Retreive controller details model.

        Returns:
            ControllerDetailModel: Controller details model
        """
        tag_filter = {
            "connection_id": self.connection_id
        }
        record: ConnectionControllerDetailsRecord = \
            await ConnectionControllerDetailsRecord.retrieve_by_tag_filter(
                context,
                tag_filter
            )
        return record.controller_details_model

    @classmethod
    async def set_connection_as_marketplace(
            cls,
            context: InjectionContext,
            connection_id: str
    ) -> "MarketplaceConnectionRecord":
        """Set connection as marketplace.

        Args:
            context (InjectionContext): Injection context to be used.
            connection_id (str): Connection identifier

        Returns:
            MarketplaceConnectionRecord: Marketplace connection record.
        """

        tag_filter = {
            "connection_id": connection_id
        }
        records = await cls.query(context, tag_filter)

        if records:
            record: MarketplaceConnectionRecord = records[0]
        else:
            # Create marketplace connection record.
            record = cls(
                connection_id=connection_id
            )

        await record.save(context)

        return record

    @classmethod
    async def retrieve_connection_record(
        cls,
        context: InjectionContext,
        connection_id: str
    ) -> ConnectionRecord:
        """Retrieve connection record.

        Args:
            context (InjectionContext): Injection context to be used.
            connection_id (str): Connection id.

        Returns:
            ConnectionRecord: Connection record.
        """
        tag_filter = {"connection_id": connection_id}
        records = await cls.query(context, tag_filter)
        assert records, "Marketplace connection not found."

        record = records[0]

        connection_record = await ConnectionRecord.retrieve_by_id(context, record.connection_id)
        return connection_record

Ancestors

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

Class variables

var Meta

var RECORD_ID_NAME

var RECORD_TYPE

var TAG_NAMES

var WEBHOOK_TOPIC

Static methods

async def retrieve_connection_record(context: aries_cloudagent.config.injection_context.InjectionContext, connection_id: str) ‑> aries_cloudagent.connections.models.connection_record.ConnectionRecord

Retrieve connection record.

Args

context : InjectionContext

Injection context to be used.

connection_id : str

Connection id.

Returns

ConnectionRecord

Connection record.

Expand source code

@classmethod
async def retrieve_connection_record(
    cls,
    context: InjectionContext,
    connection_id: str
) -> ConnectionRecord:
    """Retrieve connection record.

    Args:
        context (InjectionContext): Injection context to be used.
        connection_id (str): Connection id.

    Returns:
        ConnectionRecord: Connection record.
    """
    tag_filter = {"connection_id": connection_id}
    records = await cls.query(context, tag_filter)
    assert records, "Marketplace connection not found."

    record = records[0]

    connection_record = await ConnectionRecord.retrieve_by_id(context, record.connection_id)
    return connection_record

async def set_connection_as_marketplace(context: aries_cloudagent.config.injection_context.InjectionContext, connection_id: str) ‑> MarketplaceConnectionRecord

Set connection as marketplace.

Args

context : InjectionContext

Injection context to be used.

connection_id : str

Connection identifier

Returns

MarketplaceConnectionRecord

Marketplace connection record.

Expand source code

@classmethod
async def set_connection_as_marketplace(
        cls,
        context: InjectionContext,
        connection_id: str
) -> "MarketplaceConnectionRecord":
    """Set connection as marketplace.

    Args:
        context (InjectionContext): Injection context to be used.
        connection_id (str): Connection identifier

    Returns:
        MarketplaceConnectionRecord: Marketplace connection record.
    """

    tag_filter = {
        "connection_id": connection_id
    }
    records = await cls.query(context, tag_filter)

    if records:
        record: MarketplaceConnectionRecord = records[0]
    else:
        # Create marketplace connection record.
        record = cls(
            connection_id=connection_id
        )

    await record.save(context)

    return record

Instance variables

var record_value : dict

Accessor for JSON record value generated for this transaction record.

Expand source code

@property
def record_value(self) -> dict:
    """Accessor for JSON record value generated for this transaction record."""
    return {
        prop: getattr(self, prop)
        for prop in (
            "connection_id",
            "state"
        )
    }

Methods

async def controller_details_model(self, context: aries_cloudagent.config.injection_context.InjectionContext) ‑> ControllerDetailModel

Retreive controller details model.

Returns

ControllerDetailModel

Controller details model

Expand source code

async def controller_details_model(
    self,
    context: InjectionContext
) -> ControllerDetailModel:
    """Retreive controller details model.

    Returns:
        ControllerDetailModel: Controller details model
    """
    tag_filter = {
        "connection_id": self.connection_id
    }
    record: ConnectionControllerDetailsRecord = \
        await ConnectionControllerDetailsRecord.retrieve_by_tag_filter(
            context,
            tag_filter
        )
    return record.controller_details_model

class MarketplaceConnectionRecordSchema (*args, **kwargs)

Marketplace connection record schema

Initialize BaseModelSchema.

Raises

TypeError

If model_class is not set on Meta

Expand source code

class MarketplaceConnectionRecordSchema(BaseRecordSchema):
    """Marketplace connection record schema"""

    class Meta:
        model_class = MarketplaceConnectionRecord

    connection_id = fields.Str()

Ancestors

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

Class variables

var Meta

var opts