Module dexa_sdk.agreements.dda.v1_0.records.dda_instance_record
Expand source code
import uuid
import typing
from loguru import logger
from marshmallow import fields, validate, EXCLUDE
from aries_cloudagent.messaging.models.base_record import BaseRecord, BaseRecordSchema
from aries_cloudagent.messaging.valid import UUIDFour
from aries_cloudagent.config.injection_context import InjectionContext
from aries_cloudagent.connections.models.connection_record import ConnectionRecord
from aries_cloudagent.wallet.base import BaseWallet
from aries_cloudagent.wallet.indy import IndyWallet
from mydata_did.v1_0.utils.util import (
bool_to_str,
str_to_bool,
current_datetime_in_iso8601
)
from dexa_protocol.v1_0.messages.negotiation.offer_dda import OfferDDAMessage
from dexa_protocol.v1_0.messages.negotiation.accept_dda import AcceptDDAMessage
from dexa_protocol.v1_0.models.offer_dda_model import (
CustomerIdentificationModel
)
from .dda_template_record import DataDisclosureAgreementTemplateRecord
from .....data_controller.records.connection_controller_details_record import (
ConnectionControllerDetailsRecord
)
from ..models.dda_instance_models import (
DataUsingServiceModel,
DataDisclosureAgreementInstanceModel
)
from .....jsonld.core import sign_agreement, verify_agreement
class DataDisclosureAgreementInstanceRecord(BaseRecord):
"""Data disclosure agreement instance record to be persisted in the storage"""
class Meta:
# Schema class
schema_class = "DataDisclosureAgreementInstanceRecordSchema"
# Record type
RECORD_TYPE = "data_disclosure_agreement_instance"
# Record identifier
RECORD_ID_NAME = "id"
# Webhook topic name for this record type
WEBHOOK_TOPIC = None
# Record tags
TAG_NAMES = {
"~instance_id",
"~template_id",
"~template_version",
"~industry_sector",
"~delete_flag",
"~mydata_did",
"~blink",
"~connection_id"
}
# States of the data disclosure agreement instance.
STATE_DEFINITION = "DEFINITION"
STATE_PREPARATION = "PREPARATION"
STATE_CAPTURE = "CAPTURE"
STATE_PROOF = "PROOF"
def __init__(
self,
*,
id: str = None,
instance_id: str = None,
template_id: str = None,
template_version: str = None,
state: str = None,
data_disclosure_agreement: dict = None,
industry_sector: str = None,
delete_flag: str = "false",
connection_id: str = None,
mydata_did: str = None,
blink: str = None,
blockchain_receipt: dict = None,
customer_identification: dict = None,
**kwargs
):
"""Initialise data disclosure agreement instance record.
Args:
id (str, optional): Record ID. Defaults to None.
instance_id (str, optional): Instance ID. Defaults to None.
template_id (str, optional): Template ID. Defaults to None.
template_version (str, optional): Template version. Defaults to None.
state (str, optional): State of the record. Defaults to None.
data_disclosure_agreement (dict, optional): DDA. Defaults to None.
industry_sector (str, optional): Industry sector. Defaults to None.
delete_flag (str, optional): Delete flag. Defaults to "false".
connection_id (str, optional): Connection ID. Defaults to None.
mydata_did (str, optional): MyData DID. Defaults to None.
blink (str, optional): Blockchain Link. Defaults to None.
blockchain_receipt (dict, optional): Blockchain Receipt. Defaults to None.
customer_identification (dict, optional): Customer identification. Defauls to None.
"""
# Pass identifier and state to parent class
super().__init__(id, state, **kwargs)
if not instance_id:
raise TypeError(
"Instance identifier is not specified."
)
if not template_id:
raise TypeError(
"Template identifier is not specified."
)
if not template_version:
raise TypeError(
"Template version is not specified."
)
# Set the record attributes
self.instance_id = instance_id
self.template_id = template_id
self.template_version = template_version
self.state = state
self.data_disclosure_agreement = data_disclosure_agreement
self.industry_sector = industry_sector
self.delete_flag = delete_flag
self.connection_id = connection_id
self.mydata_did = mydata_did
self.blink = blink
self.blockchain_receipt = blockchain_receipt
self.customer_identification = customer_identification
@property
def record_value(self) -> dict:
"""Accessor for JSON record value generated for this transaction record."""
return {
prop: getattr(self, prop)
for prop in (
"instance_id",
"template_id",
"template_version",
"state",
"data_disclosure_agreement",
"industry_sector",
"delete_flag",
"connection_id",
"mydata_did",
"blink",
"blockchain_receipt",
"customer_identification"
)
}
@property
def _delete_flag(self) -> bool:
"""Accessor for delete_flag."""
return str_to_bool(self.delete_flag)
@_delete_flag.setter
def _delete_flag(self, value: bool) -> None:
"""Setter for delete_flag."""
self.delete_flag = bool_to_str(value)
@property
def is_deleted(self) -> bool:
"""Check if data agreemnent is deleted."""
return self._delete_flag
@property
def customer_identification_model(
self
) -> CustomerIdentificationModel:
return CustomerIdentificationModel.deserialize(
self.customer_identification
)
@staticmethod
async def build_instance_from_template(
context: InjectionContext,
template_id: str,
connection_record: ConnectionRecord
) -> typing.Tuple[
"DataDisclosureAgreementInstanceRecord",
DataDisclosureAgreementInstanceModel
]:
"""Build instance from DDA template.
Args:
context (InjectionContext): Injection context to be used.
template_id (str): Template ID
connection_record (ConnectionRecord): Connection record.
Returns:
DataDisclosureAgreementInstanceRecord: DDA instance record.
"""
# Fetch latest template record.
template_record = \
await DataDisclosureAgreementTemplateRecord.latest_published_template_by_id(
context,
template_id
)
assert template_record, "DDA template not found."
# Sign the template.
# Fetch wallet from context
wallet: IndyWallet = await context.inject(BaseWallet)
# Controller did (Public did)
controller_did = await wallet.get_public_did()
# Signature options
signature_options = {
"id": f"did:sov:{controller_did.did}#1",
"type": "Ed25519Signature2018",
"created": current_datetime_in_iso8601(),
"verificationMethod": f"{controller_did.verkey}",
"proofPurpose": "authentication",
}
# DDA
dda_template_dict = template_record.data_disclosure_agreement
dda_template_model = template_record.dda_model
# Add versions and ids
instance_id = str(uuid.uuid4())
instance_version = "1.0.0"
template_id = dda_template_model.id
template_version = dda_template_model.version
dda_template_dict.update({"@id": instance_id})
dda_template_dict.update({"version": instance_version})
dda_template_dict.update({"templateId": template_id})
dda_template_dict.update({"templateVersion": template_version})
# Add DUS info.
# Fetch controller details for the DUS connection
tag_filter = {
"connection_id": connection_record.connection_id
}
dus_connection_controller_details_record: ConnectionControllerDetailsRecord = \
await ConnectionControllerDetailsRecord.retrieve_by_tag_filter(
context,
tag_filter
)
data_controller_model = dus_connection_controller_details_record.controller_details_model
dus_model = DataUsingServiceModel(
did=data_controller_model.organisation_did,
name=data_controller_model.organisation_name,
url=data_controller_model.policy_url,
legal_id=data_controller_model.organisation_did,
industry_sector=data_controller_model.organisation_type,
usage_purposes=dda_template_model.purpose,
jurisdiction=dda_template_model.data_sharing_restrictions.jurisdiction,
withdrawal=data_controller_model.policy_url,
privacy_rights=data_controller_model.policy_url,
signature_contact=data_controller_model.organisation_did
)
dda_template_dict.update({"dataUsingService": dus_model.serialize()})
# Sign the DDA
signed_dda = await sign_agreement(
agreement=dda_template_dict,
verkey=controller_did.verkey,
wallet=wallet,
signature_options=signature_options
)
dda_instance_model: DataDisclosureAgreementInstanceModel = \
DataDisclosureAgreementInstanceModel.deserialize(
signed_dda
)
# Create and save DDA instance record.
dda_instance_record = DataDisclosureAgreementInstanceRecord(
instance_id=instance_id,
template_id=template_id,
template_version=template_version,
state=DataDisclosureAgreementInstanceRecord.STATE_DEFINITION,
data_disclosure_agreement=dda_instance_model.serialize(),
industry_sector=template_record.industry_sector,
connection_id=connection_record.connection_id
)
await dda_instance_record.save(context)
return (dda_instance_record, dda_instance_model)
@classmethod
async def build_instance_from_dda_offer(
cls,
context: InjectionContext,
dda_offer_message: OfferDDAMessage,
connection_record: ConnectionRecord
) -> typing.Tuple[
"DataDisclosureAgreementInstanceRecord",
DataDisclosureAgreementInstanceModel
]:
"""Build instance from DDA offer.
Args:
dda_offer_message (OfferDDAMessage): DDA offer message.
connection_record (ConnectionRecord): Connection record.
Returns:
DataDisclosureAgreementInstanceRecord: DDA instance record.
"""
# Fetch wallet from context.
wallet: IndyWallet = await context.inject(BaseWallet)
# DUS public did
dus_did = await wallet.get_public_did()
customer_identification_details = dda_offer_message.body.customer_identification
dda_offer = dda_offer_message.body.dda
dda_offer_dict = dda_offer.serialize()
# Verify the dda offer.
valid = await verify_agreement(
agreement=dda_offer_dict.copy(),
wallet=wallet
)
assert valid, "DDA instance verification failed."
# Signature options
signature_options = {
"id": f"did:sov:{dus_did.did}#1",
"type": "Ed25519Signature2018",
"created": current_datetime_in_iso8601(),
"verificationMethod": f"{dus_did.verkey}",
"proofPurpose": "authentication",
}
# Sign the DDA
signed_dda = await sign_agreement(
agreement=dda_offer_dict,
verkey=dus_did.verkey,
wallet=wallet,
signature_options=signature_options
)
dda_instance_model: DataDisclosureAgreementInstanceModel = \
DataDisclosureAgreementInstanceModel.deserialize(
signed_dda
)
# Create and save DDA instance record.
dda_instance_record = DataDisclosureAgreementInstanceRecord(
instance_id=dda_instance_model.id,
template_id=dda_instance_model.template_id,
template_version=dda_instance_model.template_version,
state=DataDisclosureAgreementInstanceRecord.STATE_CAPTURE,
data_disclosure_agreement=dda_instance_model.serialize(),
industry_sector=dda_instance_model.data_using_service.industry_sector,
connection_id=connection_record.connection_id,
customer_identification=customer_identification_details.serialize()
)
await dda_instance_record.save(context)
return (dda_instance_record, dda_instance_model)
@classmethod
async def update_instance_from_dda_accept(
cls,
context: InjectionContext,
dda_accept_message: AcceptDDAMessage
) -> "DataDisclosureAgreementInstanceRecord":
"""Update instance from DDA accept.
Args:
context (InjectionContext): Injection context to be used.
dda_accept_message (AcceptDDAMessage): DDA accept message.
connection_record (ConnectionRecord): Connection record.
Returns:
DataDisclosureAgreementInstanceRecord: DDA instance record.
"""
# Fetch wallet from context.
wallet: IndyWallet = await context.inject(BaseWallet)
dda_instance_model = dda_accept_message.body.dda
dda_accept_dict = dda_instance_model.serialize()
# Fetch instance record.
tag_filter = {
"instance_id": dda_instance_model.id
}
instance_record: DataDisclosureAgreementInstanceRecord = \
await DataDisclosureAgreementInstanceRecord.retrieve_by_tag_filter(
context,
tag_filter
)
# Verify the dda accept.
valid = await verify_agreement(
agreement=dda_accept_dict.copy(),
wallet=wallet
)
assert valid, "DDA instance verification failed."
instance_record.data_disclosure_agreement = dda_accept_dict
instance_record.state = DataDisclosureAgreementInstanceRecord.STATE_CAPTURE
await instance_record.save(context)
return instance_record
class DataDisclosureAgreementInstanceRecordSchema(BaseRecordSchema):
"""Data agreement instance record schema"""
class Meta:
# Model class
model_class = DataDisclosureAgreementInstanceRecord
# Unknown fields are excluded
unknown = EXCLUDE
# DDA instance identifier
instance_id = fields.Str(
required=True
)
# Data disclosure agreement template identifier
template_id = fields.Str(
required=True,
example=UUIDFour.EXAMPLE
)
# Data disclosure agreement template version
template_version = fields.Str(
required=False
)
# State of the data agreement.
state = fields.Str(
required=True,
example=DataDisclosureAgreementInstanceRecord.STATE_PREPARATION,
validate=validate.OneOf(
[
DataDisclosureAgreementInstanceRecord.STATE_DEFINITION,
DataDisclosureAgreementInstanceRecord.STATE_PREPARATION,
]
)
)
# Data disclosure agreement
data_disclosure_agreement = fields.Dict(required=True)
# Industry sector
industry_sector = fields.Str(required=False)
# Is deleted or not
delete_flag = fields.Str(
required=True,
example="false",
validate=validate.OneOf(
[
"true",
"false",
]
)
)
# Connection identifier
connection_id = fields.Str(required=False)
# did:mydata identifier
mydata_did = fields.Str(required=False)
# Blockchain link
blink = fields.Str(required=False)
# Blockchain receipt
blockchain_receipt = fields.Dict(required=False)
Classes
class DataDisclosureAgreementInstanceRecord (*, id: str = None, instance_id: str = None, template_id: str = None, template_version: str = None, state: str = None, data_disclosure_agreement: dict = None, industry_sector: str = None, delete_flag: str = 'false', connection_id: str = None, mydata_did: str = None, blink: str = None, blockchain_receipt: dict = None, customer_identification: dict = None, **kwargs)
-
Data disclosure agreement instance record to be persisted in the storage
Initialise data disclosure agreement instance record.
Args
id
:str
, optional- Record ID. Defaults to None.
instance_id
:str
, optional- Instance ID. Defaults to None.
template_id
:str
, optional- Template ID. Defaults to None.
template_version
:str
, optional- Template version. Defaults to None.
state
:str
, optional- State of the record. Defaults to None.
data_disclosure_agreement
:dict
, optional- DDA. Defaults to None.
industry_sector
:str
, optional- Industry sector. Defaults to None.
delete_flag
:str
, optional- Delete flag. Defaults to "false".
connection_id
:str
, optional- Connection ID. Defaults to None.
mydata_did
:str
, optional- MyData DID. Defaults to None.
blink
:str
, optional- Blockchain Link. Defaults to None.
blockchain_receipt
:dict
, optional- Blockchain Receipt. Defaults to None.
customer_identification
:dict
, optional- Customer identification. Defauls to None.
Expand source code
class DataDisclosureAgreementInstanceRecord(BaseRecord): """Data disclosure agreement instance record to be persisted in the storage""" class Meta: # Schema class schema_class = "DataDisclosureAgreementInstanceRecordSchema" # Record type RECORD_TYPE = "data_disclosure_agreement_instance" # Record identifier RECORD_ID_NAME = "id" # Webhook topic name for this record type WEBHOOK_TOPIC = None # Record tags TAG_NAMES = { "~instance_id", "~template_id", "~template_version", "~industry_sector", "~delete_flag", "~mydata_did", "~blink", "~connection_id" } # States of the data disclosure agreement instance. STATE_DEFINITION = "DEFINITION" STATE_PREPARATION = "PREPARATION" STATE_CAPTURE = "CAPTURE" STATE_PROOF = "PROOF" def __init__( self, *, id: str = None, instance_id: str = None, template_id: str = None, template_version: str = None, state: str = None, data_disclosure_agreement: dict = None, industry_sector: str = None, delete_flag: str = "false", connection_id: str = None, mydata_did: str = None, blink: str = None, blockchain_receipt: dict = None, customer_identification: dict = None, **kwargs ): """Initialise data disclosure agreement instance record. Args: id (str, optional): Record ID. Defaults to None. instance_id (str, optional): Instance ID. Defaults to None. template_id (str, optional): Template ID. Defaults to None. template_version (str, optional): Template version. Defaults to None. state (str, optional): State of the record. Defaults to None. data_disclosure_agreement (dict, optional): DDA. Defaults to None. industry_sector (str, optional): Industry sector. Defaults to None. delete_flag (str, optional): Delete flag. Defaults to "false". connection_id (str, optional): Connection ID. Defaults to None. mydata_did (str, optional): MyData DID. Defaults to None. blink (str, optional): Blockchain Link. Defaults to None. blockchain_receipt (dict, optional): Blockchain Receipt. Defaults to None. customer_identification (dict, optional): Customer identification. Defauls to None. """ # Pass identifier and state to parent class super().__init__(id, state, **kwargs) if not instance_id: raise TypeError( "Instance identifier is not specified." ) if not template_id: raise TypeError( "Template identifier is not specified." ) if not template_version: raise TypeError( "Template version is not specified." ) # Set the record attributes self.instance_id = instance_id self.template_id = template_id self.template_version = template_version self.state = state self.data_disclosure_agreement = data_disclosure_agreement self.industry_sector = industry_sector self.delete_flag = delete_flag self.connection_id = connection_id self.mydata_did = mydata_did self.blink = blink self.blockchain_receipt = blockchain_receipt self.customer_identification = customer_identification @property def record_value(self) -> dict: """Accessor for JSON record value generated for this transaction record.""" return { prop: getattr(self, prop) for prop in ( "instance_id", "template_id", "template_version", "state", "data_disclosure_agreement", "industry_sector", "delete_flag", "connection_id", "mydata_did", "blink", "blockchain_receipt", "customer_identification" ) } @property def _delete_flag(self) -> bool: """Accessor for delete_flag.""" return str_to_bool(self.delete_flag) @_delete_flag.setter def _delete_flag(self, value: bool) -> None: """Setter for delete_flag.""" self.delete_flag = bool_to_str(value) @property def is_deleted(self) -> bool: """Check if data agreemnent is deleted.""" return self._delete_flag @property def customer_identification_model( self ) -> CustomerIdentificationModel: return CustomerIdentificationModel.deserialize( self.customer_identification ) @staticmethod async def build_instance_from_template( context: InjectionContext, template_id: str, connection_record: ConnectionRecord ) -> typing.Tuple[ "DataDisclosureAgreementInstanceRecord", DataDisclosureAgreementInstanceModel ]: """Build instance from DDA template. Args: context (InjectionContext): Injection context to be used. template_id (str): Template ID connection_record (ConnectionRecord): Connection record. Returns: DataDisclosureAgreementInstanceRecord: DDA instance record. """ # Fetch latest template record. template_record = \ await DataDisclosureAgreementTemplateRecord.latest_published_template_by_id( context, template_id ) assert template_record, "DDA template not found." # Sign the template. # Fetch wallet from context wallet: IndyWallet = await context.inject(BaseWallet) # Controller did (Public did) controller_did = await wallet.get_public_did() # Signature options signature_options = { "id": f"did:sov:{controller_did.did}#1", "type": "Ed25519Signature2018", "created": current_datetime_in_iso8601(), "verificationMethod": f"{controller_did.verkey}", "proofPurpose": "authentication", } # DDA dda_template_dict = template_record.data_disclosure_agreement dda_template_model = template_record.dda_model # Add versions and ids instance_id = str(uuid.uuid4()) instance_version = "1.0.0" template_id = dda_template_model.id template_version = dda_template_model.version dda_template_dict.update({"@id": instance_id}) dda_template_dict.update({"version": instance_version}) dda_template_dict.update({"templateId": template_id}) dda_template_dict.update({"templateVersion": template_version}) # Add DUS info. # Fetch controller details for the DUS connection tag_filter = { "connection_id": connection_record.connection_id } dus_connection_controller_details_record: ConnectionControllerDetailsRecord = \ await ConnectionControllerDetailsRecord.retrieve_by_tag_filter( context, tag_filter ) data_controller_model = dus_connection_controller_details_record.controller_details_model dus_model = DataUsingServiceModel( did=data_controller_model.organisation_did, name=data_controller_model.organisation_name, url=data_controller_model.policy_url, legal_id=data_controller_model.organisation_did, industry_sector=data_controller_model.organisation_type, usage_purposes=dda_template_model.purpose, jurisdiction=dda_template_model.data_sharing_restrictions.jurisdiction, withdrawal=data_controller_model.policy_url, privacy_rights=data_controller_model.policy_url, signature_contact=data_controller_model.organisation_did ) dda_template_dict.update({"dataUsingService": dus_model.serialize()}) # Sign the DDA signed_dda = await sign_agreement( agreement=dda_template_dict, verkey=controller_did.verkey, wallet=wallet, signature_options=signature_options ) dda_instance_model: DataDisclosureAgreementInstanceModel = \ DataDisclosureAgreementInstanceModel.deserialize( signed_dda ) # Create and save DDA instance record. dda_instance_record = DataDisclosureAgreementInstanceRecord( instance_id=instance_id, template_id=template_id, template_version=template_version, state=DataDisclosureAgreementInstanceRecord.STATE_DEFINITION, data_disclosure_agreement=dda_instance_model.serialize(), industry_sector=template_record.industry_sector, connection_id=connection_record.connection_id ) await dda_instance_record.save(context) return (dda_instance_record, dda_instance_model) @classmethod async def build_instance_from_dda_offer( cls, context: InjectionContext, dda_offer_message: OfferDDAMessage, connection_record: ConnectionRecord ) -> typing.Tuple[ "DataDisclosureAgreementInstanceRecord", DataDisclosureAgreementInstanceModel ]: """Build instance from DDA offer. Args: dda_offer_message (OfferDDAMessage): DDA offer message. connection_record (ConnectionRecord): Connection record. Returns: DataDisclosureAgreementInstanceRecord: DDA instance record. """ # Fetch wallet from context. wallet: IndyWallet = await context.inject(BaseWallet) # DUS public did dus_did = await wallet.get_public_did() customer_identification_details = dda_offer_message.body.customer_identification dda_offer = dda_offer_message.body.dda dda_offer_dict = dda_offer.serialize() # Verify the dda offer. valid = await verify_agreement( agreement=dda_offer_dict.copy(), wallet=wallet ) assert valid, "DDA instance verification failed." # Signature options signature_options = { "id": f"did:sov:{dus_did.did}#1", "type": "Ed25519Signature2018", "created": current_datetime_in_iso8601(), "verificationMethod": f"{dus_did.verkey}", "proofPurpose": "authentication", } # Sign the DDA signed_dda = await sign_agreement( agreement=dda_offer_dict, verkey=dus_did.verkey, wallet=wallet, signature_options=signature_options ) dda_instance_model: DataDisclosureAgreementInstanceModel = \ DataDisclosureAgreementInstanceModel.deserialize( signed_dda ) # Create and save DDA instance record. dda_instance_record = DataDisclosureAgreementInstanceRecord( instance_id=dda_instance_model.id, template_id=dda_instance_model.template_id, template_version=dda_instance_model.template_version, state=DataDisclosureAgreementInstanceRecord.STATE_CAPTURE, data_disclosure_agreement=dda_instance_model.serialize(), industry_sector=dda_instance_model.data_using_service.industry_sector, connection_id=connection_record.connection_id, customer_identification=customer_identification_details.serialize() ) await dda_instance_record.save(context) return (dda_instance_record, dda_instance_model) @classmethod async def update_instance_from_dda_accept( cls, context: InjectionContext, dda_accept_message: AcceptDDAMessage ) -> "DataDisclosureAgreementInstanceRecord": """Update instance from DDA accept. Args: context (InjectionContext): Injection context to be used. dda_accept_message (AcceptDDAMessage): DDA accept message. connection_record (ConnectionRecord): Connection record. Returns: DataDisclosureAgreementInstanceRecord: DDA instance record. """ # Fetch wallet from context. wallet: IndyWallet = await context.inject(BaseWallet) dda_instance_model = dda_accept_message.body.dda dda_accept_dict = dda_instance_model.serialize() # Fetch instance record. tag_filter = { "instance_id": dda_instance_model.id } instance_record: DataDisclosureAgreementInstanceRecord = \ await DataDisclosureAgreementInstanceRecord.retrieve_by_tag_filter( context, tag_filter ) # Verify the dda accept. valid = await verify_agreement( agreement=dda_accept_dict.copy(), wallet=wallet ) assert valid, "DDA instance verification failed." instance_record.data_disclosure_agreement = dda_accept_dict instance_record.state = DataDisclosureAgreementInstanceRecord.STATE_CAPTURE await instance_record.save(context) return instance_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 STATE_CAPTURE
var STATE_DEFINITION
var STATE_PREPARATION
var STATE_PROOF
var TAG_NAMES
var WEBHOOK_TOPIC
Static methods
async def build_instance_from_dda_offer(context: aries_cloudagent.config.injection_context.InjectionContext, dda_offer_message: dexa_protocol.v1_0.messages.negotiation.offer_dda.OfferDDAMessage, connection_record: aries_cloudagent.connections.models.connection_record.ConnectionRecord) ‑> Tuple[DataDisclosureAgreementInstanceRecord, DataDisclosureAgreementInstanceModel]
-
Build instance from DDA offer.
Args
dda_offer_message
:OfferDDAMessage
- DDA offer message.
connection_record
:ConnectionRecord
- Connection record.
Returns
DataDisclosureAgreementInstanceRecord
- DDA instance record.
Expand source code
@classmethod async def build_instance_from_dda_offer( cls, context: InjectionContext, dda_offer_message: OfferDDAMessage, connection_record: ConnectionRecord ) -> typing.Tuple[ "DataDisclosureAgreementInstanceRecord", DataDisclosureAgreementInstanceModel ]: """Build instance from DDA offer. Args: dda_offer_message (OfferDDAMessage): DDA offer message. connection_record (ConnectionRecord): Connection record. Returns: DataDisclosureAgreementInstanceRecord: DDA instance record. """ # Fetch wallet from context. wallet: IndyWallet = await context.inject(BaseWallet) # DUS public did dus_did = await wallet.get_public_did() customer_identification_details = dda_offer_message.body.customer_identification dda_offer = dda_offer_message.body.dda dda_offer_dict = dda_offer.serialize() # Verify the dda offer. valid = await verify_agreement( agreement=dda_offer_dict.copy(), wallet=wallet ) assert valid, "DDA instance verification failed." # Signature options signature_options = { "id": f"did:sov:{dus_did.did}#1", "type": "Ed25519Signature2018", "created": current_datetime_in_iso8601(), "verificationMethod": f"{dus_did.verkey}", "proofPurpose": "authentication", } # Sign the DDA signed_dda = await sign_agreement( agreement=dda_offer_dict, verkey=dus_did.verkey, wallet=wallet, signature_options=signature_options ) dda_instance_model: DataDisclosureAgreementInstanceModel = \ DataDisclosureAgreementInstanceModel.deserialize( signed_dda ) # Create and save DDA instance record. dda_instance_record = DataDisclosureAgreementInstanceRecord( instance_id=dda_instance_model.id, template_id=dda_instance_model.template_id, template_version=dda_instance_model.template_version, state=DataDisclosureAgreementInstanceRecord.STATE_CAPTURE, data_disclosure_agreement=dda_instance_model.serialize(), industry_sector=dda_instance_model.data_using_service.industry_sector, connection_id=connection_record.connection_id, customer_identification=customer_identification_details.serialize() ) await dda_instance_record.save(context) return (dda_instance_record, dda_instance_model)
async def build_instance_from_template(context: aries_cloudagent.config.injection_context.InjectionContext, template_id: str, connection_record: aries_cloudagent.connections.models.connection_record.ConnectionRecord) ‑> Tuple[DataDisclosureAgreementInstanceRecord, DataDisclosureAgreementInstanceModel]
-
Build instance from DDA template.
Args
context
:InjectionContext
- Injection context to be used.
template_id
:str
- Template ID
connection_record
:ConnectionRecord
- Connection record.
Returns
DataDisclosureAgreementInstanceRecord
- DDA instance record.
Expand source code
@staticmethod async def build_instance_from_template( context: InjectionContext, template_id: str, connection_record: ConnectionRecord ) -> typing.Tuple[ "DataDisclosureAgreementInstanceRecord", DataDisclosureAgreementInstanceModel ]: """Build instance from DDA template. Args: context (InjectionContext): Injection context to be used. template_id (str): Template ID connection_record (ConnectionRecord): Connection record. Returns: DataDisclosureAgreementInstanceRecord: DDA instance record. """ # Fetch latest template record. template_record = \ await DataDisclosureAgreementTemplateRecord.latest_published_template_by_id( context, template_id ) assert template_record, "DDA template not found." # Sign the template. # Fetch wallet from context wallet: IndyWallet = await context.inject(BaseWallet) # Controller did (Public did) controller_did = await wallet.get_public_did() # Signature options signature_options = { "id": f"did:sov:{controller_did.did}#1", "type": "Ed25519Signature2018", "created": current_datetime_in_iso8601(), "verificationMethod": f"{controller_did.verkey}", "proofPurpose": "authentication", } # DDA dda_template_dict = template_record.data_disclosure_agreement dda_template_model = template_record.dda_model # Add versions and ids instance_id = str(uuid.uuid4()) instance_version = "1.0.0" template_id = dda_template_model.id template_version = dda_template_model.version dda_template_dict.update({"@id": instance_id}) dda_template_dict.update({"version": instance_version}) dda_template_dict.update({"templateId": template_id}) dda_template_dict.update({"templateVersion": template_version}) # Add DUS info. # Fetch controller details for the DUS connection tag_filter = { "connection_id": connection_record.connection_id } dus_connection_controller_details_record: ConnectionControllerDetailsRecord = \ await ConnectionControllerDetailsRecord.retrieve_by_tag_filter( context, tag_filter ) data_controller_model = dus_connection_controller_details_record.controller_details_model dus_model = DataUsingServiceModel( did=data_controller_model.organisation_did, name=data_controller_model.organisation_name, url=data_controller_model.policy_url, legal_id=data_controller_model.organisation_did, industry_sector=data_controller_model.organisation_type, usage_purposes=dda_template_model.purpose, jurisdiction=dda_template_model.data_sharing_restrictions.jurisdiction, withdrawal=data_controller_model.policy_url, privacy_rights=data_controller_model.policy_url, signature_contact=data_controller_model.organisation_did ) dda_template_dict.update({"dataUsingService": dus_model.serialize()}) # Sign the DDA signed_dda = await sign_agreement( agreement=dda_template_dict, verkey=controller_did.verkey, wallet=wallet, signature_options=signature_options ) dda_instance_model: DataDisclosureAgreementInstanceModel = \ DataDisclosureAgreementInstanceModel.deserialize( signed_dda ) # Create and save DDA instance record. dda_instance_record = DataDisclosureAgreementInstanceRecord( instance_id=instance_id, template_id=template_id, template_version=template_version, state=DataDisclosureAgreementInstanceRecord.STATE_DEFINITION, data_disclosure_agreement=dda_instance_model.serialize(), industry_sector=template_record.industry_sector, connection_id=connection_record.connection_id ) await dda_instance_record.save(context) return (dda_instance_record, dda_instance_model)
async def update_instance_from_dda_accept(context: aries_cloudagent.config.injection_context.InjectionContext, dda_accept_message: dexa_protocol.v1_0.messages.negotiation.accept_dda.AcceptDDAMessage) ‑> DataDisclosureAgreementInstanceRecord
-
Update instance from DDA accept.
Args
context
:InjectionContext
- Injection context to be used.
dda_accept_message
:AcceptDDAMessage
- DDA accept message.
connection_record
:ConnectionRecord
- Connection record.
Returns
DataDisclosureAgreementInstanceRecord
- DDA instance record.
Expand source code
@classmethod async def update_instance_from_dda_accept( cls, context: InjectionContext, dda_accept_message: AcceptDDAMessage ) -> "DataDisclosureAgreementInstanceRecord": """Update instance from DDA accept. Args: context (InjectionContext): Injection context to be used. dda_accept_message (AcceptDDAMessage): DDA accept message. connection_record (ConnectionRecord): Connection record. Returns: DataDisclosureAgreementInstanceRecord: DDA instance record. """ # Fetch wallet from context. wallet: IndyWallet = await context.inject(BaseWallet) dda_instance_model = dda_accept_message.body.dda dda_accept_dict = dda_instance_model.serialize() # Fetch instance record. tag_filter = { "instance_id": dda_instance_model.id } instance_record: DataDisclosureAgreementInstanceRecord = \ await DataDisclosureAgreementInstanceRecord.retrieve_by_tag_filter( context, tag_filter ) # Verify the dda accept. valid = await verify_agreement( agreement=dda_accept_dict.copy(), wallet=wallet ) assert valid, "DDA instance verification failed." instance_record.data_disclosure_agreement = dda_accept_dict instance_record.state = DataDisclosureAgreementInstanceRecord.STATE_CAPTURE await instance_record.save(context) return instance_record
Instance variables
var customer_identification_model : dexa_protocol.v1_0.models.offer_dda_model.CustomerIdentificationModel
-
Expand source code
@property def customer_identification_model( self ) -> CustomerIdentificationModel: return CustomerIdentificationModel.deserialize( self.customer_identification )
var is_deleted : bool
-
Check if data agreemnent is deleted.
Expand source code
@property def is_deleted(self) -> bool: """Check if data agreemnent is deleted.""" return self._delete_flag
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 ( "instance_id", "template_id", "template_version", "state", "data_disclosure_agreement", "industry_sector", "delete_flag", "connection_id", "mydata_did", "blink", "blockchain_receipt", "customer_identification" ) }
class DataDisclosureAgreementInstanceRecordSchema (*args, **kwargs)
-
Data agreement instance record schema
Initialize BaseModelSchema.
Raises
TypeError
- If model_class is not set on Meta
Expand source code
class DataDisclosureAgreementInstanceRecordSchema(BaseRecordSchema): """Data agreement instance record schema""" class Meta: # Model class model_class = DataDisclosureAgreementInstanceRecord # Unknown fields are excluded unknown = EXCLUDE # DDA instance identifier instance_id = fields.Str( required=True ) # Data disclosure agreement template identifier template_id = fields.Str( required=True, example=UUIDFour.EXAMPLE ) # Data disclosure agreement template version template_version = fields.Str( required=False ) # State of the data agreement. state = fields.Str( required=True, example=DataDisclosureAgreementInstanceRecord.STATE_PREPARATION, validate=validate.OneOf( [ DataDisclosureAgreementInstanceRecord.STATE_DEFINITION, DataDisclosureAgreementInstanceRecord.STATE_PREPARATION, ] ) ) # Data disclosure agreement data_disclosure_agreement = fields.Dict(required=True) # Industry sector industry_sector = fields.Str(required=False) # Is deleted or not delete_flag = fields.Str( required=True, example="false", validate=validate.OneOf( [ "true", "false", ] ) ) # Connection identifier connection_id = fields.Str(required=False) # did:mydata identifier mydata_did = fields.Str(required=False) # Blockchain link blink = fields.Str(required=False) # Blockchain receipt blockchain_receipt = fields.Dict(required=False)
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