Module dexa_sdk.agreements.da.v1_0.records.da_template_record
Expand source code
import typing
from marshmallow import fields, validate, EXCLUDE
from loguru import logger
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 mydata_did.v1_0.utils.util import bool_to_str, str_to_bool
from .....utils import bump_major_for_semver_string
from .personal_data_record import PersonalDataRecord
from ..models.da_models import DataAgreementModel
class DataAgreementTemplateRecord(BaseRecord):
"""Data agreement template record to be persisted in the storage"""
class Meta:
# Schema class
schema_class = "DataAgreementTemplateRecordSchema"
# Record type
RECORD_TYPE = "data_agreement_template"
# Record identifier
RECORD_ID_NAME = "id"
# Webhook topic name for this record type
WEBHOOK_TOPIC = None
# Record tags
TAG_NAMES = {
"~template_id",
"~template_version",
"~method_of_use",
"~publish_flag",
"~delete_flag",
"~schema_id",
"~cred_def_id",
"~existing_schema_flag",
"~latest_version_flag",
"~third_party_data_sharing"
}
# States of the data agreement.
STATE_DEFINITION = "DEFINITION"
STATE_PREPARATION = "PREPARATION"
STATE_CAPTURE = "CAPTURE"
STATE_PROOF = "PROOF"
METHOD_OF_USE_DATA_SOURCE = "data-source"
METHOD_OF_USE_DATA_USING_SERVICE = "data-using-service"
def __init__(
self,
*,
id: str = None,
template_id: str = None,
template_version: str = None,
state: str = None,
method_of_use: str = None,
data_agreement: dict = None,
publish_flag: str = "false",
delete_flag: str = "false",
schema_id: str = None,
cred_def_id: str = None,
presentation_request: dict = None,
existing_schema_flag: str = "false",
latest_version_flag: str = "true",
third_party_data_sharing: str = "false",
**kwargs
):
"""Initialise data agreement template record
Args:
id (str, optional): Identifier of record. Defaults to None.
template_id (str, optional): Identifier for the template. Default to None.
template_version (str, optional): Version for the template. Default to None.
state (str, optional): State of the data agreeement template. Defaults to None.
method_of_use (str, optional): Method of use. Defaults to None.
data_agreement (dict, optional): Data agreement template. Defaults to None.
publish_flag (str, optional): Published or not. Defaults to "false".
delete_flag (str, optional): Deleted or not. Defaults to "false".
schema_id (str, optional): Schema identifier. Defaults to None.
cred_def_id (str, optional): Credential definition identifier. Defaults to None.
presentation_request (dict, optional): Presentation request. Defaults to None.
existing_schema_flag (str, optional): Is existing schema or not. Defaults to "false".
latest_version_flag (str, optional): Latest version of record or not.
Defaults to "false".
"""
# Pass identifier and state to parent class
super().__init__(id, state, **kwargs)
if not template_id:
raise TypeError(
"Template identifier is not specified."
)
# Set the record attributes
self.template_id = template_id
self.template_version = template_version
self.method_of_use = method_of_use
self.state = state
self.data_agreement = data_agreement
self.publish_flag = publish_flag
self.delete_flag = delete_flag
self.schema_id = schema_id
self.cred_def_id = cred_def_id
self.presentation_request = presentation_request
self.existing_schema_flag = existing_schema_flag
self.latest_version_flag = latest_version_flag
self.third_party_data_sharing = third_party_data_sharing
@property
def record_value(self) -> dict:
"""Accessor for JSON record value generated for this transaction record."""
return {
prop: getattr(self, prop)
for prop in (
"template_id",
"template_version",
"method_of_use",
"state",
"data_agreement",
"publish_flag",
"delete_flag",
"schema_id",
"cred_def_id",
"presentation_request",
"existing_schema_flag",
"latest_version_flag",
"third_party_data_sharing"
)
}
@property
def _publish_flag(self) -> bool:
"""Accessor for publish_flag."""
return str_to_bool(self.publish_flag)
@_publish_flag.setter
def _publish_flag(self, value: bool) -> None:
"""Setter for publish_flag."""
self.publish_flag = bool_to_str(value)
@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 _existing_schema_flag(self) -> bool:
"""Accessor for existing_schema_flag."""
return str_to_bool(self.existing_schema_flag)
@_existing_schema_flag.setter
def _existing_schema_flag(self, value: bool) -> None:
"""Setter for existing_schema_flag."""
self.existing_schema_flag = bool_to_str(value)
@property
def _latest_version_flag(self) -> bool:
"""Accessor for latest_version_flag."""
return str_to_bool(self.latest_version_flag)
@_latest_version_flag.setter
def _latest_version_flag(self, value: bool) -> None:
"""Setter for latest_version_flag."""
self.latest_version_flag = bool_to_str(value)
@property
def is_published(self) -> bool:
"""Check if data agreement record is published."""
return self._publish_flag
@property
def is_deleted(self) -> bool:
"""Check if data agreemnent is deleted."""
return self._delete_flag
@property
def is_draft(self) -> bool:
"""Check if data agreement is a draft."""
return True if not self._publish_flag else False
@property
def is_existing_schema(self) -> bool:
"""Check if the schema provided was existing or not."""
return self._existing_schema_flag
@property
def is_latest_version(self) -> bool:
"""Check if the current record is latest version."""
return self._latest_version_flag
@classmethod
async def latest_published_template_by_id(
cls,
context: InjectionContext,
template_id: str,
) -> "DataAgreementTemplateRecord":
"""Fetch latest published template by id.
Args:
context (InjectionContext): Injection context to use.
template_id (str): Data agreement template id.
Returns:
DataAgreementTemplateRecord: Data agreement template record.
"""
tag_filter: dict = {
"delete_flag": bool_to_str(False),
"template_id": template_id,
"latest_version_flag": bool_to_str(True),
"publish_flag": bool_to_str(True)
}
fetched = await cls.query(
context,
tag_filter=tag_filter
)
return None if len(fetched) == 0 else fetched[0]
@classmethod
async def latest_template_by_id(
cls,
context: InjectionContext,
template_id: str,
) -> "DataAgreementTemplateRecord":
"""Fetch latest template by id.
Args:
context (InjectionContext): Injection context to use.
template_id (str): Data agreement template id.
Returns:
DataAgreementTemplateRecord: Data agreement template record.
"""
tag_filter: dict = {
"delete_flag": bool_to_str(False),
"template_id": template_id,
"latest_version_flag": bool_to_str(True)
}
fetched = await cls.query(
context,
tag_filter=tag_filter
)
return None if len(fetched) == 0 else fetched[0]
@classmethod
async def non_deleted_templates_by_id(
cls,
context: InjectionContext,
template_id: str,
) -> typing.List["DataAgreementTemplateRecord"]:
"""Fetch a non-deleted templates by id.
Args:
context (InjectionContext): Injection context to use.
template_id (str): Data agreement template id.
Returns:
DataAgreementTemplateRecord: Data agreement template record.
"""
tag_filter: dict = {
"delete_flag": bool_to_str(False),
"template_id": template_id
}
fetched = await cls.query(
context,
tag_filter=tag_filter
)
return fetched
@classmethod
async def non_deleted_template_by_id(
cls,
context: InjectionContext,
template_id: str
) -> typing.List["DataAgreementTemplateRecord"]:
"""Fetch non deleted template by id.
Returns:
DataAgreementTemplateRecord: Template record
"""
tag_filter: dict = {
"latest_version_flag": bool_to_str(True),
"delete_flag": bool_to_str(False),
"template_id": template_id
}
return await cls.query(
context,
tag_filter=tag_filter,
)
@classmethod
async def non_deleted_templates(
cls,
context: InjectionContext
) -> typing.List["DataAgreementTemplateRecord"]:
"""Fetch all non-deleted agreements.
Returns:
DataAgreementTemplateRecord: List of template records
"""
tag_filter: dict = {
"delete_flag": bool_to_str(False)
}
return await cls.query(
context,
tag_filter=tag_filter,
)
async def upgrade(self, context: InjectionContext, **kwargs) -> str:
"""Upgrade the data agreement template to next version.
Args:
context (InjectionContext): Injection context to be used.
Returns:
str: Returns the record identifier of the upgraded record.
"""
# Fetch the previous record
previous_record: DataAgreementTemplateRecord = \
await DataAgreementTemplateRecord.latest_template_by_id(
context,
self.template_id
)
assert previous_record, "Atleast 1 previous record must be present"
# Mark the previous record as not the latest
previous_record._latest_version_flag = False
await previous_record.save(context)
# Bump the version of data agreement.
previous_data_agreement = previous_record.data_agreement
previous_version = previous_data_agreement["version"]
template_version = bump_major_for_semver_string(previous_version)
logger.info(
(
f"Data agreement template version is bumped from"
f" {previous_version} to {template_version}"
)
)
self.data_agreement.update(
{
"version": template_version
}
)
# Save as new record to the storage
self._id = None
self.template_version = template_version
await self.save(context=context, **kwargs)
return self
@classmethod
async def published_templates(cls, context: InjectionContext) \
-> typing.List["DataAgreementTemplateRecord"]:
"""Fetch all published templates (not-deleted)
Returns:
DataAgreementTemplateRecord: List of template records
"""
tag_filter: dict = {
"delete_flag": bool_to_str(False),
"publish_flag": bool_to_str(True)
}
return await cls.query(
context,
tag_filter=tag_filter
)
async def publish_template(self, context: InjectionContext, **kwargs) -> str:
"""Publish the data agreement template
Args:
context (InjectionContext): Injection context to be used.
Returns:
str: Returns the record identifier for the published record.
"""
self._publish_flag = True
return await self.save(context=context, **kwargs)
async def delete_template(self, context: InjectionContext, **kwargs) -> str:
"""Delete the data agreement template
Args:
context (InjectionContext): Injection context to be used.
Returns:
str: Returns the record identifier for deleted record.
"""
self._delete_flag = True
return await self.save(context=context, **kwargs)
async def fetch_personal_data_records(
self,
context: InjectionContext
) -> typing.List[PersonalDataRecord]:
"""Fetch personal data records
Args:
context (InjectionContext): Injection context to be used.
Returns:
typing.List[PersonalDataRecord]: Personal data records
"""
logger.info(
(
f"Fetching personal data records for DA template with "
f"id:{self.template_id} and version:{self.template_version}"
)
)
return await PersonalDataRecord.list_by_template_id(
context,
self.template_id,
self.template_version
)
@property
def data_agreement_model(self) -> DataAgreementModel:
"""Accessor for data agreement as model.
"""
return DataAgreementModel.deserialize(self.data_agreement)
class DataAgreementTemplateRecordSchema(BaseRecordSchema):
"""Data agreement template record schema"""
class Meta:
# Model class
model_class = DataAgreementTemplateRecord
# Unknown fields are excluded
unknown = EXCLUDE
# Data agreement template identifier
template_id = fields.Str(
required=True,
example=UUIDFour.EXAMPLE
)
# Data agreement template version
template_version = fields.Str(
required=False
)
# State of the data agreement.
state = fields.Str(
required=True,
example=DataAgreementTemplateRecord.STATE_PREPARATION,
validate=validate.OneOf(
[
DataAgreementTemplateRecord.STATE_DEFINITION,
DataAgreementTemplateRecord.STATE_PREPARATION,
]
)
)
# Method of use for the data agreement.
method_of_use = fields.Str(
required=True,
example="data-source",
validate=validate.OneOf(
[
DataAgreementTemplateRecord.METHOD_OF_USE_DATA_SOURCE,
DataAgreementTemplateRecord.METHOD_OF_USE_DATA_USING_SERVICE,
]
)
)
# Data agreement
data_agreement = fields.Dict(
required=True
)
# Schema identifier
schema_id = fields.Str(
required=True,
example="WgWxqztrNooG92RXvxSTWv:2:schema_name:1.0"
)
# Credential definition identifier
cred_def_id = fields.Str(
required=True,
example="WgWxqztrNooG92RXvxSTWv:3:CL:20:tag"
)
# Presentation request
presentation_request = fields.Dict(
required=True,
)
# Is published or not
publish_flag = fields.Str(
required=True,
example="false",
validate=validate.OneOf(
[
"true",
"false",
]
)
)
# Is deleted or not
delete_flag = fields.Str(
required=True,
example="false",
validate=validate.OneOf(
[
"true",
"false",
]
)
)
# Is existing schema or not
existing_schema_flag = fields.Str(
required=True,
example="false",
validate=validate.OneOf(
[
"true",
"false",
]
)
)
# Latest version of the record or not.
latest_version_flag = fields.Str(
required=True,
example="false",
validate=validate.OneOf(
[
"true",
"false",
]
)
)
# Third party data sharing
third_party_data_sharing = fields.Str(
required=True,
example="false",
validate=validate.OneOf(
[
"true",
"false",
]
)
)
Classes
class DataAgreementTemplateRecord (*, id: str = None, template_id: str = None, template_version: str = None, state: str = None, method_of_use: str = None, data_agreement: dict = None, publish_flag: str = 'false', delete_flag: str = 'false', schema_id: str = None, cred_def_id: str = None, presentation_request: dict = None, existing_schema_flag: str = 'false', latest_version_flag: str = 'true', third_party_data_sharing: str = 'false', **kwargs)
-
Data agreement template record to be persisted in the storage
Initialise data agreement template record
Args
id
:str
, optional- Identifier of record. Defaults to None.
template_id
:str
, optional- Identifier for the template. Default to None.
template_version
:str
, optional- Version for the template. Default to None.
state
:str
, optional- State of the data agreeement template. Defaults to None.
method_of_use
:str
, optional- Method of use. Defaults to None.
data_agreement
:dict
, optional- Data agreement template. Defaults to None.
publish_flag
:str
, optional- Published or not. Defaults to "false".
delete_flag
:str
, optional- Deleted or not. Defaults to "false".
schema_id
:str
, optional- Schema identifier. Defaults to None.
cred_def_id
:str
, optional- Credential definition identifier. Defaults to None.
presentation_request
:dict
, optional- Presentation request. Defaults to None.
existing_schema_flag
:str
, optional- Is existing schema or not. Defaults to "false".
latest_version_flag
:str
, optional- Latest version of record or not. Defaults to "false".
Expand source code
class DataAgreementTemplateRecord(BaseRecord): """Data agreement template record to be persisted in the storage""" class Meta: # Schema class schema_class = "DataAgreementTemplateRecordSchema" # Record type RECORD_TYPE = "data_agreement_template" # Record identifier RECORD_ID_NAME = "id" # Webhook topic name for this record type WEBHOOK_TOPIC = None # Record tags TAG_NAMES = { "~template_id", "~template_version", "~method_of_use", "~publish_flag", "~delete_flag", "~schema_id", "~cred_def_id", "~existing_schema_flag", "~latest_version_flag", "~third_party_data_sharing" } # States of the data agreement. STATE_DEFINITION = "DEFINITION" STATE_PREPARATION = "PREPARATION" STATE_CAPTURE = "CAPTURE" STATE_PROOF = "PROOF" METHOD_OF_USE_DATA_SOURCE = "data-source" METHOD_OF_USE_DATA_USING_SERVICE = "data-using-service" def __init__( self, *, id: str = None, template_id: str = None, template_version: str = None, state: str = None, method_of_use: str = None, data_agreement: dict = None, publish_flag: str = "false", delete_flag: str = "false", schema_id: str = None, cred_def_id: str = None, presentation_request: dict = None, existing_schema_flag: str = "false", latest_version_flag: str = "true", third_party_data_sharing: str = "false", **kwargs ): """Initialise data agreement template record Args: id (str, optional): Identifier of record. Defaults to None. template_id (str, optional): Identifier for the template. Default to None. template_version (str, optional): Version for the template. Default to None. state (str, optional): State of the data agreeement template. Defaults to None. method_of_use (str, optional): Method of use. Defaults to None. data_agreement (dict, optional): Data agreement template. Defaults to None. publish_flag (str, optional): Published or not. Defaults to "false". delete_flag (str, optional): Deleted or not. Defaults to "false". schema_id (str, optional): Schema identifier. Defaults to None. cred_def_id (str, optional): Credential definition identifier. Defaults to None. presentation_request (dict, optional): Presentation request. Defaults to None. existing_schema_flag (str, optional): Is existing schema or not. Defaults to "false". latest_version_flag (str, optional): Latest version of record or not. Defaults to "false". """ # Pass identifier and state to parent class super().__init__(id, state, **kwargs) if not template_id: raise TypeError( "Template identifier is not specified." ) # Set the record attributes self.template_id = template_id self.template_version = template_version self.method_of_use = method_of_use self.state = state self.data_agreement = data_agreement self.publish_flag = publish_flag self.delete_flag = delete_flag self.schema_id = schema_id self.cred_def_id = cred_def_id self.presentation_request = presentation_request self.existing_schema_flag = existing_schema_flag self.latest_version_flag = latest_version_flag self.third_party_data_sharing = third_party_data_sharing @property def record_value(self) -> dict: """Accessor for JSON record value generated for this transaction record.""" return { prop: getattr(self, prop) for prop in ( "template_id", "template_version", "method_of_use", "state", "data_agreement", "publish_flag", "delete_flag", "schema_id", "cred_def_id", "presentation_request", "existing_schema_flag", "latest_version_flag", "third_party_data_sharing" ) } @property def _publish_flag(self) -> bool: """Accessor for publish_flag.""" return str_to_bool(self.publish_flag) @_publish_flag.setter def _publish_flag(self, value: bool) -> None: """Setter for publish_flag.""" self.publish_flag = bool_to_str(value) @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 _existing_schema_flag(self) -> bool: """Accessor for existing_schema_flag.""" return str_to_bool(self.existing_schema_flag) @_existing_schema_flag.setter def _existing_schema_flag(self, value: bool) -> None: """Setter for existing_schema_flag.""" self.existing_schema_flag = bool_to_str(value) @property def _latest_version_flag(self) -> bool: """Accessor for latest_version_flag.""" return str_to_bool(self.latest_version_flag) @_latest_version_flag.setter def _latest_version_flag(self, value: bool) -> None: """Setter for latest_version_flag.""" self.latest_version_flag = bool_to_str(value) @property def is_published(self) -> bool: """Check if data agreement record is published.""" return self._publish_flag @property def is_deleted(self) -> bool: """Check if data agreemnent is deleted.""" return self._delete_flag @property def is_draft(self) -> bool: """Check if data agreement is a draft.""" return True if not self._publish_flag else False @property def is_existing_schema(self) -> bool: """Check if the schema provided was existing or not.""" return self._existing_schema_flag @property def is_latest_version(self) -> bool: """Check if the current record is latest version.""" return self._latest_version_flag @classmethod async def latest_published_template_by_id( cls, context: InjectionContext, template_id: str, ) -> "DataAgreementTemplateRecord": """Fetch latest published template by id. Args: context (InjectionContext): Injection context to use. template_id (str): Data agreement template id. Returns: DataAgreementTemplateRecord: Data agreement template record. """ tag_filter: dict = { "delete_flag": bool_to_str(False), "template_id": template_id, "latest_version_flag": bool_to_str(True), "publish_flag": bool_to_str(True) } fetched = await cls.query( context, tag_filter=tag_filter ) return None if len(fetched) == 0 else fetched[0] @classmethod async def latest_template_by_id( cls, context: InjectionContext, template_id: str, ) -> "DataAgreementTemplateRecord": """Fetch latest template by id. Args: context (InjectionContext): Injection context to use. template_id (str): Data agreement template id. Returns: DataAgreementTemplateRecord: Data agreement template record. """ tag_filter: dict = { "delete_flag": bool_to_str(False), "template_id": template_id, "latest_version_flag": bool_to_str(True) } fetched = await cls.query( context, tag_filter=tag_filter ) return None if len(fetched) == 0 else fetched[0] @classmethod async def non_deleted_templates_by_id( cls, context: InjectionContext, template_id: str, ) -> typing.List["DataAgreementTemplateRecord"]: """Fetch a non-deleted templates by id. Args: context (InjectionContext): Injection context to use. template_id (str): Data agreement template id. Returns: DataAgreementTemplateRecord: Data agreement template record. """ tag_filter: dict = { "delete_flag": bool_to_str(False), "template_id": template_id } fetched = await cls.query( context, tag_filter=tag_filter ) return fetched @classmethod async def non_deleted_template_by_id( cls, context: InjectionContext, template_id: str ) -> typing.List["DataAgreementTemplateRecord"]: """Fetch non deleted template by id. Returns: DataAgreementTemplateRecord: Template record """ tag_filter: dict = { "latest_version_flag": bool_to_str(True), "delete_flag": bool_to_str(False), "template_id": template_id } return await cls.query( context, tag_filter=tag_filter, ) @classmethod async def non_deleted_templates( cls, context: InjectionContext ) -> typing.List["DataAgreementTemplateRecord"]: """Fetch all non-deleted agreements. Returns: DataAgreementTemplateRecord: List of template records """ tag_filter: dict = { "delete_flag": bool_to_str(False) } return await cls.query( context, tag_filter=tag_filter, ) async def upgrade(self, context: InjectionContext, **kwargs) -> str: """Upgrade the data agreement template to next version. Args: context (InjectionContext): Injection context to be used. Returns: str: Returns the record identifier of the upgraded record. """ # Fetch the previous record previous_record: DataAgreementTemplateRecord = \ await DataAgreementTemplateRecord.latest_template_by_id( context, self.template_id ) assert previous_record, "Atleast 1 previous record must be present" # Mark the previous record as not the latest previous_record._latest_version_flag = False await previous_record.save(context) # Bump the version of data agreement. previous_data_agreement = previous_record.data_agreement previous_version = previous_data_agreement["version"] template_version = bump_major_for_semver_string(previous_version) logger.info( ( f"Data agreement template version is bumped from" f" {previous_version} to {template_version}" ) ) self.data_agreement.update( { "version": template_version } ) # Save as new record to the storage self._id = None self.template_version = template_version await self.save(context=context, **kwargs) return self @classmethod async def published_templates(cls, context: InjectionContext) \ -> typing.List["DataAgreementTemplateRecord"]: """Fetch all published templates (not-deleted) Returns: DataAgreementTemplateRecord: List of template records """ tag_filter: dict = { "delete_flag": bool_to_str(False), "publish_flag": bool_to_str(True) } return await cls.query( context, tag_filter=tag_filter ) async def publish_template(self, context: InjectionContext, **kwargs) -> str: """Publish the data agreement template Args: context (InjectionContext): Injection context to be used. Returns: str: Returns the record identifier for the published record. """ self._publish_flag = True return await self.save(context=context, **kwargs) async def delete_template(self, context: InjectionContext, **kwargs) -> str: """Delete the data agreement template Args: context (InjectionContext): Injection context to be used. Returns: str: Returns the record identifier for deleted record. """ self._delete_flag = True return await self.save(context=context, **kwargs) async def fetch_personal_data_records( self, context: InjectionContext ) -> typing.List[PersonalDataRecord]: """Fetch personal data records Args: context (InjectionContext): Injection context to be used. Returns: typing.List[PersonalDataRecord]: Personal data records """ logger.info( ( f"Fetching personal data records for DA template with " f"id:{self.template_id} and version:{self.template_version}" ) ) return await PersonalDataRecord.list_by_template_id( context, self.template_id, self.template_version ) @property def data_agreement_model(self) -> DataAgreementModel: """Accessor for data agreement as model. """ return DataAgreementModel.deserialize(self.data_agreement)
Ancestors
- aries_cloudagent.messaging.models.base_record.BaseRecord
- aries_cloudagent.messaging.models.base.BaseModel
- abc.ABC
Class variables
var METHOD_OF_USE_DATA_SOURCE
var METHOD_OF_USE_DATA_USING_SERVICE
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 latest_published_template_by_id(context: aries_cloudagent.config.injection_context.InjectionContext, template_id: str) ‑> DataAgreementTemplateRecord
-
Fetch latest published template by id.
Args
context
:InjectionContext
- Injection context to use.
template_id
:str
- Data agreement template id.
Returns
DataAgreementTemplateRecord
- Data agreement template record.
Expand source code
@classmethod async def latest_published_template_by_id( cls, context: InjectionContext, template_id: str, ) -> "DataAgreementTemplateRecord": """Fetch latest published template by id. Args: context (InjectionContext): Injection context to use. template_id (str): Data agreement template id. Returns: DataAgreementTemplateRecord: Data agreement template record. """ tag_filter: dict = { "delete_flag": bool_to_str(False), "template_id": template_id, "latest_version_flag": bool_to_str(True), "publish_flag": bool_to_str(True) } fetched = await cls.query( context, tag_filter=tag_filter ) return None if len(fetched) == 0 else fetched[0]
async def latest_template_by_id(context: aries_cloudagent.config.injection_context.InjectionContext, template_id: str) ‑> DataAgreementTemplateRecord
-
Fetch latest template by id.
Args
context
:InjectionContext
- Injection context to use.
template_id
:str
- Data agreement template id.
Returns
DataAgreementTemplateRecord
- Data agreement template record.
Expand source code
@classmethod async def latest_template_by_id( cls, context: InjectionContext, template_id: str, ) -> "DataAgreementTemplateRecord": """Fetch latest template by id. Args: context (InjectionContext): Injection context to use. template_id (str): Data agreement template id. Returns: DataAgreementTemplateRecord: Data agreement template record. """ tag_filter: dict = { "delete_flag": bool_to_str(False), "template_id": template_id, "latest_version_flag": bool_to_str(True) } fetched = await cls.query( context, tag_filter=tag_filter ) return None if len(fetched) == 0 else fetched[0]
async def non_deleted_template_by_id(context: aries_cloudagent.config.injection_context.InjectionContext, template_id: str) ‑> List[DataAgreementTemplateRecord]
-
Expand source code
@classmethod async def non_deleted_template_by_id( cls, context: InjectionContext, template_id: str ) -> typing.List["DataAgreementTemplateRecord"]: """Fetch non deleted template by id. Returns: DataAgreementTemplateRecord: Template record """ tag_filter: dict = { "latest_version_flag": bool_to_str(True), "delete_flag": bool_to_str(False), "template_id": template_id } return await cls.query( context, tag_filter=tag_filter, )
async def non_deleted_templates(context: aries_cloudagent.config.injection_context.InjectionContext) ‑> List[DataAgreementTemplateRecord]
-
Expand source code
@classmethod async def non_deleted_templates( cls, context: InjectionContext ) -> typing.List["DataAgreementTemplateRecord"]: """Fetch all non-deleted agreements. Returns: DataAgreementTemplateRecord: List of template records """ tag_filter: dict = { "delete_flag": bool_to_str(False) } return await cls.query( context, tag_filter=tag_filter, )
async def non_deleted_templates_by_id(context: aries_cloudagent.config.injection_context.InjectionContext, template_id: str) ‑> List[DataAgreementTemplateRecord]
-
Fetch a non-deleted templates by id.
Args
context
:InjectionContext
- Injection context to use.
template_id
:str
- Data agreement template id.
Returns
DataAgreementTemplateRecord
- Data agreement template record.
Expand source code
@classmethod async def non_deleted_templates_by_id( cls, context: InjectionContext, template_id: str, ) -> typing.List["DataAgreementTemplateRecord"]: """Fetch a non-deleted templates by id. Args: context (InjectionContext): Injection context to use. template_id (str): Data agreement template id. Returns: DataAgreementTemplateRecord: Data agreement template record. """ tag_filter: dict = { "delete_flag": bool_to_str(False), "template_id": template_id } fetched = await cls.query( context, tag_filter=tag_filter ) return fetched
async def published_templates(context: aries_cloudagent.config.injection_context.InjectionContext) ‑> List[DataAgreementTemplateRecord]
-
Fetch all published templates (not-deleted)
Returns
DataAgreementTemplateRecord
- List of template records
Expand source code
@classmethod async def published_templates(cls, context: InjectionContext) \ -> typing.List["DataAgreementTemplateRecord"]: """Fetch all published templates (not-deleted) Returns: DataAgreementTemplateRecord: List of template records """ tag_filter: dict = { "delete_flag": bool_to_str(False), "publish_flag": bool_to_str(True) } return await cls.query( context, tag_filter=tag_filter )
Instance variables
var data_agreement_model : DataAgreementModel
-
Accessor for data agreement as model.
Expand source code
@property def data_agreement_model(self) -> DataAgreementModel: """Accessor for data agreement as model. """ return DataAgreementModel.deserialize(self.data_agreement)
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 is_draft : bool
-
Check if data agreement is a draft.
Expand source code
@property def is_draft(self) -> bool: """Check if data agreement is a draft.""" return True if not self._publish_flag else False
var is_existing_schema : bool
-
Check if the schema provided was existing or not.
Expand source code
@property def is_existing_schema(self) -> bool: """Check if the schema provided was existing or not.""" return self._existing_schema_flag
var is_latest_version : bool
-
Check if the current record is latest version.
Expand source code
@property def is_latest_version(self) -> bool: """Check if the current record is latest version.""" return self._latest_version_flag
var is_published : bool
-
Check if data agreement record is published.
Expand source code
@property def is_published(self) -> bool: """Check if data agreement record is published.""" return self._publish_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 ( "template_id", "template_version", "method_of_use", "state", "data_agreement", "publish_flag", "delete_flag", "schema_id", "cred_def_id", "presentation_request", "existing_schema_flag", "latest_version_flag", "third_party_data_sharing" ) }
Methods
async def delete_template(self, context: aries_cloudagent.config.injection_context.InjectionContext, **kwargs) ‑> str
-
Delete the data agreement template
Args
context
:InjectionContext
- Injection context to be used.
Returns
str
- Returns the record identifier for deleted record.
Expand source code
async def delete_template(self, context: InjectionContext, **kwargs) -> str: """Delete the data agreement template Args: context (InjectionContext): Injection context to be used. Returns: str: Returns the record identifier for deleted record. """ self._delete_flag = True return await self.save(context=context, **kwargs)
async def fetch_personal_data_records(self, context: aries_cloudagent.config.injection_context.InjectionContext) ‑> List[PersonalDataRecord]
-
Fetch personal data records
Args
context
:InjectionContext
- Injection context to be used.
Returns
typing.List[PersonalDataRecord]
- Personal data records
Expand source code
async def fetch_personal_data_records( self, context: InjectionContext ) -> typing.List[PersonalDataRecord]: """Fetch personal data records Args: context (InjectionContext): Injection context to be used. Returns: typing.List[PersonalDataRecord]: Personal data records """ logger.info( ( f"Fetching personal data records for DA template with " f"id:{self.template_id} and version:{self.template_version}" ) ) return await PersonalDataRecord.list_by_template_id( context, self.template_id, self.template_version )
async def publish_template(self, context: aries_cloudagent.config.injection_context.InjectionContext, **kwargs) ‑> str
-
Publish the data agreement template
Args
context
:InjectionContext
- Injection context to be used.
Returns
str
- Returns the record identifier for the published record.
Expand source code
async def publish_template(self, context: InjectionContext, **kwargs) -> str: """Publish the data agreement template Args: context (InjectionContext): Injection context to be used. Returns: str: Returns the record identifier for the published record. """ self._publish_flag = True return await self.save(context=context, **kwargs)
async def upgrade(self, context: aries_cloudagent.config.injection_context.InjectionContext, **kwargs) ‑> str
-
Upgrade the data agreement template to next version.
Args
context
:InjectionContext
- Injection context to be used.
Returns
str
- Returns the record identifier of the upgraded record.
Expand source code
async def upgrade(self, context: InjectionContext, **kwargs) -> str: """Upgrade the data agreement template to next version. Args: context (InjectionContext): Injection context to be used. Returns: str: Returns the record identifier of the upgraded record. """ # Fetch the previous record previous_record: DataAgreementTemplateRecord = \ await DataAgreementTemplateRecord.latest_template_by_id( context, self.template_id ) assert previous_record, "Atleast 1 previous record must be present" # Mark the previous record as not the latest previous_record._latest_version_flag = False await previous_record.save(context) # Bump the version of data agreement. previous_data_agreement = previous_record.data_agreement previous_version = previous_data_agreement["version"] template_version = bump_major_for_semver_string(previous_version) logger.info( ( f"Data agreement template version is bumped from" f" {previous_version} to {template_version}" ) ) self.data_agreement.update( { "version": template_version } ) # Save as new record to the storage self._id = None self.template_version = template_version await self.save(context=context, **kwargs) return self
class DataAgreementTemplateRecordSchema (*args, **kwargs)
-
Data agreement template record schema
Initialize BaseModelSchema.
Raises
TypeError
- If model_class is not set on Meta
Expand source code
class DataAgreementTemplateRecordSchema(BaseRecordSchema): """Data agreement template record schema""" class Meta: # Model class model_class = DataAgreementTemplateRecord # Unknown fields are excluded unknown = EXCLUDE # Data agreement template identifier template_id = fields.Str( required=True, example=UUIDFour.EXAMPLE ) # Data agreement template version template_version = fields.Str( required=False ) # State of the data agreement. state = fields.Str( required=True, example=DataAgreementTemplateRecord.STATE_PREPARATION, validate=validate.OneOf( [ DataAgreementTemplateRecord.STATE_DEFINITION, DataAgreementTemplateRecord.STATE_PREPARATION, ] ) ) # Method of use for the data agreement. method_of_use = fields.Str( required=True, example="data-source", validate=validate.OneOf( [ DataAgreementTemplateRecord.METHOD_OF_USE_DATA_SOURCE, DataAgreementTemplateRecord.METHOD_OF_USE_DATA_USING_SERVICE, ] ) ) # Data agreement data_agreement = fields.Dict( required=True ) # Schema identifier schema_id = fields.Str( required=True, example="WgWxqztrNooG92RXvxSTWv:2:schema_name:1.0" ) # Credential definition identifier cred_def_id = fields.Str( required=True, example="WgWxqztrNooG92RXvxSTWv:3:CL:20:tag" ) # Presentation request presentation_request = fields.Dict( required=True, ) # Is published or not publish_flag = fields.Str( required=True, example="false", validate=validate.OneOf( [ "true", "false", ] ) ) # Is deleted or not delete_flag = fields.Str( required=True, example="false", validate=validate.OneOf( [ "true", "false", ] ) ) # Is existing schema or not existing_schema_flag = fields.Str( required=True, example="false", validate=validate.OneOf( [ "true", "false", ] ) ) # Latest version of the record or not. latest_version_flag = fields.Str( required=True, example="false", validate=validate.OneOf( [ "true", "false", ] ) ) # Third party data sharing third_party_data_sharing = fields.Str( required=True, example="false", validate=validate.OneOf( [ "true", "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