Create Event With Verified Domain

# Create an event for an asset given url to Archivist and user Token.
#
# The module contains four functions: main, create_asset, create_event and
# get_verified_domain.
#
# The main function would initialize an archivist connection using the url and
# the credentials, called "arch".

# create_asset will execute 'assets.create', which is a archivist connection function
# to create a new asset for the archivist through archivist connection. The main funciton then
# calls create_event and pass in "arch" and the created asset to create a new event for the asset.

# In both cases the verified domain name is displayed.

from json import dumps as json_dumps
from os import getenv
from warnings import filterwarnings

from archivist.archivist import Archivist
from archivist.logger import set_logger

filterwarnings("ignore", message="Unverified HTTPS request")

%reload_ext dotenv
%dotenv -o notebooks.env

# URL, CLIENT, SECRET are environment variables that represent connection parameters.
#
# URL = represents the url to the DataTrails application
# CLIENT = represents the client ID from an Application Registration
# SECRET = represents the client secret from an Application Registration
DATATRAILS_URL = getenv("DATATRAILS_URL")
DATATRAILS_APPREG_CLIENT = getenv("DATATRAILS_APPREG_CLIENT")
DATATRAILS_APPREG_SECRET = getenv("DATATRAILS_APPREG_SECRET")

"""
Main function of Asset and Event creation.

* Connect to DataTrails with client ID and client secret
* Creates an Asset and two Events
* Prints response of Asset and Event creation
"""

# Optional call to set the logger level.  The argument can be either
# "INFO" or "DEBUG".  For more sophisticated logging control see our
# documentation.
set_logger("INFO")

# Initialize connection to DATATRAILS
print("Connecting to DATATRAILS")
print("DATATRAILS_URL", DATATRAILS_URL)
arch = Archivist(
    DATATRAILS_URL, (DATATRAILS_APPREG_CLIENT, DATATRAILS_APPREG_SECRET), max_time=300
)

Connecting to DATATRAILS
DATATRAILS_URL https://app.datatrails.ai

def get_verified_domain(arch, entity):
    """Get the verified domain for the asset/event.

    Args:
        arch: archivist connection.
        entity: an asset or event

    Returns:
        verified_domain: name of the verified domain
                         for the asset or event.

    """

    tenancy = arch.tenancies.publicinfo(entity["tenant_identity"])
    return tenancy.get("verified_domain", "")

def create_event(arch, asset):
    """Create an event for the passed-in asset.

    Args:
        arch: archivist connection.
        asset: an asset created using aconn

    Returns:
        new_event: a new event for the asset.

    """

    # props can be defined for different behaviours and the attributes associated with
    # different behaviours are also different.
    props = {
        "operation": "Record",
        # This event is used to record evidence.
        "behaviour": "RecordEvidence",
        # Optional Client-claimed time at which the maintenance was performed
        "timestamp_declared": "2019-11-27T14:44:19Z",
        # Optional Client-claimed identity of person performing the operation
        "principal_declared": {
            "issuer": "idp.synsation.io/1234",
            "subject": "phil.b",
            "email": "phil.b@synsation.io",
        },
    }
    attrs = {
        # Required Details of the RecordEvidence request
        "arc_description": "Safety conformance approved for version 1.6.",
        # Required The evidence to be retained in the asset history
        "arc_evidence": "DVA Conformance Report attached",
        # Example Client can add any additional information in further attributes,
        # including free text or attachments
        "conformance_report": "blobs/e2a1d16c-03cd-45a1-8cd0-690831df1273",
    }

    return arch.events.create(asset["identity"], props=props, attrs=attrs, confirm=True)

    # alternatively if some work can be done whilst the event is confirmed then this call can be
    # replaced by a two-step alternative:

    # event = arch.events.create(asset["identity"], props=props, attrs=attrs)

    # ... do something else here
    # and then wait for confirmation

    # self.arch.events.wait_for_confirmation(event['identity'])

def create_asset(arch):
    """Create an asset using Archivist Connection.

    Args:
        arch: archivist connection.

    Returns:
        newasset: a new asset created.

    """
    attrs = {
        "arc_display_name": "display_name",  # Asset's display name in the user interface
        "arc_description": "display_description",  # Asset's description in the user interface
        "arc_display_type": "display_type",  # Arc_display_type is a free text field
        # allowing the creator of
        # an asset to specify the asset
        # type or class. Be careful when setting this:
        # assets are grouped by type and
        # sharing policies can be
        # configured to share assets based on
        # their arc_display_type.
        # So a mistake here can result in asset data being
        # under- or over-shared.
        "some_custom_attribute": "value",  # You can add any custom value as long as
        # it does not start with arc_
    }
    # The first argument is the attributes of the asset
    # The second argument is wait for confirmation:
    #   If @confirm@ is True then this function will not
    #   return until the asset is confirmed.)
    return arch.assets.create(attrs=attrs, confirm=True)

# Create a new asset
asset = create_asset(arch)
print("Asset", json_dumps(asset, sort_keys=True, indent=4))
print("Verified domain '", get_verified_domain(arch, asset), "'")

Refresh token

Asset {
    "at_time": "2023-01-16T11:53:18Z",
    "attributes": {
        "arc_description": "display_description",
        "arc_display_name": "display_name",
        "arc_display_type": "display_type",
        "some_custom_attribute": "value"
    },
    "behaviours": [
        "Builtin",
        "AssetCreator",
        "RecordEvidence"
    ],
    "chain_id": "827586838445807967",
    "confirmation_status": "CONFIRMED",
    "identity": "assets/0047715f-368e-4a62-aa04-48e6fa974e85",
    "owner": "0xe889E67FdBa658C6f27ccBDa98D9d1B5500Dbbce",
    "public": false,
    "storage_integrity": "TENANT_STORAGE",
    "tenant_identity": "tenant/9bfb80ee-81f6-40dc-b5c7-1c7fb2fb9866",
    "tracked": "TRACKED"
}
Verified domain '  '

# Create a new event
event = create_event(arch, asset)
print("Event", json_dumps(event, sort_keys=True, indent=4))
print("Verified domain '", get_verified_domain(arch, event), "'")

Event {
    "asset_attributes": {},
    "asset_identity": "assets/0047715f-368e-4a62-aa04-48e6fa974e85",
    "behaviour": "RecordEvidence",
    "block_number": 0,
    "confirmation_status": "CONFIRMED",
    "event_attributes": {
        "arc_description": "Safety conformance approved for version 1.6.",
        "arc_evidence": "DVA Conformance Report attached",
        "conformance_report": "blobs/e2a1d16c-03cd-45a1-8cd0-690831df1273"
    },
    "from": "0xe889E67FdBa658C6f27ccBDa98D9d1B5500Dbbce",
    "identity": "assets/0047715f-368e-4a62-aa04-48e6fa974e85/events/f94ae41a-fe0d-4c2c-9486-46eeb2579a1f",
    "operation": "Record",
    "principal_accepted": {
        "display_name": "Test Notebooks",
        "email": "",
        "issuer": "https://app.datatrails.ai/appidpv1",
        "subject": "437bd138-dade-4346-aadd-dfdfee51ddf4"
    },
    "principal_declared": {
        "display_name": "",
        "email": "phil.b@synsation.io",
        "issuer": "idp.synsation.io/1234",
        "subject": "phil.b"
    },
    "tenant_identity": "tenant/9bfb80ee-81f6-40dc-b5c7-1c7fb2fb9866",
    "timestamp_accepted": "2023-01-16T11:53:21Z",
    "timestamp_committed": "2023-01-16T11:53:21.760625716Z",
    "timestamp_declared": "2019-11-27T14:44:19Z",
    "transaction_id": "",
    "transaction_index": 0
}
Verified domain '  '

# Fetch the event
event = arch.events.read(event["identity"])
print("Event", json_dumps(event, sort_keys=True, indent=4))
print("Verified domain '", get_verified_domain(arch, event), "'")

Event {
    "asset_attributes": {},
    "asset_identity": "assets/0047715f-368e-4a62-aa04-48e6fa974e85",
    "behaviour": "RecordEvidence",
    "block_number": 0,
    "confirmation_status": "CONFIRMED",
    "event_attributes": {
        "arc_description": "Safety conformance approved for version 1.6.",
        "arc_evidence": "DVA Conformance Report attached",
        "conformance_report": "blobs/e2a1d16c-03cd-45a1-8cd0-690831df1273"
    },
    "from": "0xe889E67FdBa658C6f27ccBDa98D9d1B5500Dbbce",
    "identity": "assets/0047715f-368e-4a62-aa04-48e6fa974e85/events/f94ae41a-fe0d-4c2c-9486-46eeb2579a1f",
    "operation": "Record",
    "principal_accepted": {
        "display_name": "Test Notebooks",
        "email": "",
        "issuer": "https://app.datatrails.ai/appidpv1",
        "subject": "437bd138-dade-4346-aadd-dfdfee51ddf4"
    },
    "principal_declared": {
        "display_name": "",
        "email": "phil.b@synsation.io",
        "issuer": "idp.synsation.io/1234",
        "subject": "phil.b"
    },
    "tenant_identity": "tenant/9bfb80ee-81f6-40dc-b5c7-1c7fb2fb9866",
    "timestamp_accepted": "2023-01-16T11:53:21Z",
    "timestamp_committed": "2023-01-16T11:53:21.760625716Z",
    "timestamp_declared": "2019-11-27T14:44:19Z",
    "transaction_id": "",
    "transaction_index": 0
}
Verified domain '  '