Hedera DID module

This module provides API for Hedera DID management.

Based on latest Hedera DID spec.

DID resolver

HederaDidResolver

Hedera DID Resolver implementation.

Parameters:

Name	Type	Description	Default
client	Client	Hedera Client	required
cache_instance	Cache[str, TimestampedRecord[DidDocument]] | None	Custom cache instance. If not provided, in-memory cache is used	None

Source code in hiero_did_sdk_python/did/hedera_did_resolver.py

class HederaDidResolver:
    """Hedera DID Resolver implementation.

    Args:
        client: Hedera Client
        cache_instance: Custom cache instance. If not provided, in-memory cache is used
    """

    def __init__(
        self,
        client: Client,
        cache_instance: Cache[str, TimestampedRecord[DidDocument]] | None = None,
    ):
        self._client = client
        self._cache = cache_instance or MemoryCache[str, TimestampedRecord[DidDocument]]()

    async def resolve(self, did: str) -> DIDResolutionResult:
        """
        Resolve DID document by identifier.

        Args:
            did: DID identifier to resolve

        Returns:
            object: DID resolution result
        """
        try:
            parsed_identifier = parse_identifier(did)
            topic_id = parsed_identifier.topic_id

            timestamped_record: TimestampedRecord | None = self._cache.get(topic_id)

            if timestamped_record:
                now = time.time()
                last_updated_timestamp: float = timestamped_record.timestamp
                did_document: DidDocument = timestamped_record.data

                if (now - last_updated_timestamp) > INSERTION_THRESHOLD_SECONDS:
                    result = await HcsMessageResolver(
                        topic_id,
                        HcsDidMessageEnvelope,
                        timestamp_from=Timestamp(int(last_updated_timestamp), 0),
                    ).execute(self._client)

                    messages = [
                        cast(HcsDidMessage, envelope.message) for envelope in cast(list[HcsDidMessageEnvelope], result)
                    ]

                    await did_document.process_messages(messages)

                    self._cache.set(
                        topic_id,
                        TimestampedRecord(did_document, did_document.updated or did_document.created or time.time()),
                    )
            else:
                registered_did = HederaDid(identifier=did, client=self._client)

                did_document = await registered_did.resolve()

                self._cache.set(
                    topic_id,
                    TimestampedRecord(did_document, did_document.updated or did_document.created or time.time()),
                )

            document_meta = {
                "versionId": did_document.version_id,
            }

            if not did_document.deactivated:
                document_meta.update({
                    "created": datetime.date.fromtimestamp(cast(float, did_document.created)).isoformat(),
                    "updated": datetime.date.fromtimestamp(cast(float, did_document.updated)).isoformat(),
                })

            status = {"deactivated": True} if did_document.deactivated else {}

            return {
                "didDocumentMetadata": cast(DIDDocumentMetadata, {**status, **document_meta}),
                "didResolutionMetadata": {"contentType": "application/did+ld+json"},
                "didDocument": cast(DIDDocument, did_document.get_json_payload()),
            }
        except Exception as error:
            return {
                "didResolutionMetadata": {
                    "error": _get_error_description(error),
                    "message": str(error),  # pyright: ignore - this is not in spec, but may be helpful
                },
                "didDocumentMetadata": {},
                "didDocument": None,
            }

resolve(did) async

Resolve DID document by identifier.

Parameters:

Name	Type	Description	Default
did	str	DID identifier to resolve	required

Returns:

Name	Type	Description
object	DIDResolutionResult	DID resolution result

Source code in hiero_did_sdk_python/did/hedera_did_resolver.py

async def resolve(self, did: str) -> DIDResolutionResult:
    """
    Resolve DID document by identifier.

    Args:
        did: DID identifier to resolve

    Returns:
        object: DID resolution result
    """
    try:
        parsed_identifier = parse_identifier(did)
        topic_id = parsed_identifier.topic_id

        timestamped_record: TimestampedRecord | None = self._cache.get(topic_id)

        if timestamped_record:
            now = time.time()
            last_updated_timestamp: float = timestamped_record.timestamp
            did_document: DidDocument = timestamped_record.data

            if (now - last_updated_timestamp) > INSERTION_THRESHOLD_SECONDS:
                result = await HcsMessageResolver(
                    topic_id,
                    HcsDidMessageEnvelope,
                    timestamp_from=Timestamp(int(last_updated_timestamp), 0),
                ).execute(self._client)

                messages = [
                    cast(HcsDidMessage, envelope.message) for envelope in cast(list[HcsDidMessageEnvelope], result)
                ]

                await did_document.process_messages(messages)

                self._cache.set(
                    topic_id,
                    TimestampedRecord(did_document, did_document.updated or did_document.created or time.time()),
                )
        else:
            registered_did = HederaDid(identifier=did, client=self._client)

            did_document = await registered_did.resolve()

            self._cache.set(
                topic_id,
                TimestampedRecord(did_document, did_document.updated or did_document.created or time.time()),
            )

        document_meta = {
            "versionId": did_document.version_id,
        }

        if not did_document.deactivated:
            document_meta.update({
                "created": datetime.date.fromtimestamp(cast(float, did_document.created)).isoformat(),
                "updated": datetime.date.fromtimestamp(cast(float, did_document.updated)).isoformat(),
            })

        status = {"deactivated": True} if did_document.deactivated else {}

        return {
            "didDocumentMetadata": cast(DIDDocumentMetadata, {**status, **document_meta}),
            "didResolutionMetadata": {"contentType": "application/did+ld+json"},
            "didDocument": cast(DIDDocument, did_document.get_json_payload()),
        }
    except Exception as error:
        return {
            "didResolutionMetadata": {
                "error": _get_error_description(error),
                "message": str(error),  # pyright: ignore - this is not in spec, but may be helpful
            },
            "didDocumentMetadata": {},
            "didDocument": None,
        }

DidResolutionError

Bases: StrEnum

Enum for DID resolution errors

Source code in hiero_did_sdk_python/did/hedera_did_resolver.py

class DidResolutionError(StrEnum):
    """Enum for DID resolution errors"""

    """
    The resolver has failed to construct the DID document.
    This can be caused by a network issue, a wrong registry address or malformed logs while parsing the registry history.
    Please inspect the `DIDResolutionMetadata.message` to debug further.
    """
    NOT_FOUND = "notFound"

    """
    The resolver does not know how to resolve the given DID. Most likely it is not a `did:hedera`.
    """
    INVALID_DID = "invalidDid"

    """
    The resolver is misconfigured or is being asked to resolve a DID anchored on an unknown network
    """
    UNKNOWN_NETWORK = "unknownNetwork"

    """
    Unknown resolution error
    """
    UNKNOWN = "unknown"

INVALID_DID = 'invalidDid' class-attribute instance-attribute

The resolver is misconfigured or is being asked to resolve a DID anchored on an unknown network

NOT_FOUND = 'notFound' class-attribute instance-attribute

The resolver does not know how to resolve the given DID. Most likely it is not a did:hedera.

UNKNOWN_NETWORK = 'unknownNetwork' class-attribute instance-attribute

Unknown resolution error

DID management API

HederaDid

Class representing Hedera DID instance, provides access to DID management API.

Parameters:

Name	Type	Description	Default
client	Client	Hedera Client	required
identifier	str | None	DID identifier (for existing DIDs)	None
private_key_der	str | None	DID Owner (controller) private key encoded in DER format. Can be empty for read-only access	None

Source code in hiero_did_sdk_python/did/hedera_did.py

class HederaDid:
    """
    Class representing Hedera DID instance, provides access to DID management API.

    Args:
        client: Hedera Client
        identifier: DID identifier (for existing DIDs)
        private_key_der: DID Owner (controller) private key encoded in DER format. Can be empty for read-only access
    """

    def __init__(self, client: Client, identifier: str | None = None, private_key_der: str | None = None):
        if not identifier and not private_key_der:
            raise DidException("'identifier' and 'private_key_der' cannot both be empty")

        self._client = client
        self._hcs_topic_service = HcsTopicService(client)

        self._private_key = PrivateKey.from_string(private_key_der) if private_key_der else None
        self._key_type: SupportedKeyType | None = (
            cast(SupportedKeyType, get_key_type(self._private_key)) if self._private_key else None
        )

        self.identifier = identifier
        if self.identifier:
            parsed_identifier = parse_identifier(self.identifier)
            self.network = parsed_identifier.network
            self.topic_id = parsed_identifier.topic_id
        else:
            self.topic_id = None

        self._messages: list[HcsDidMessage] = []
        self.document: DidDocument | None = None

    async def register(self):
        """Register (create) DID instance in Hedera network"""
        if not self._private_key or not self._key_type:
            raise DidException("Private key is required to register new DID")

        if self.identifier:
            document = await self.resolve()
            if document.controller:
                raise DidException("DID is already registered")
        else:
            topic_options = HcsTopicOptions(
                admin_key=self._private_key.public_key(), submit_key=self._private_key.public_key()
            )

            self.topic_id = await self._hcs_topic_service.create_topic(topic_options, [self._private_key])

            self.network = self._client.network.network
            self.identifier = build_identifier(
                self.network,
                multibase_encode(bytes(self._private_key.public_key().to_bytes_raw()), "base58btc"),
                self.topic_id,
            )

        hcs_event = HcsDidUpdateDidOwnerEvent(
            id_=f"{self.identifier}#did-root-key",
            controller=self.identifier,
            public_key=self._private_key.public_key(),
            type_=self._key_type,
        )

        await self._submit_transaction(DidDocumentOperation.CREATE, hcs_event)

    async def change_owner(self, controller: str, new_private_key_der: str):
        """
        Change DID Owner (controller).

        Args:
            controller: Identifier of new DID Owner
            new_private_key_der: New DID Owner private key encoded in DER format
        """
        self._assert_can_submit_transaction()

        document = await self.resolve()
        if not document.controller:
            raise DidException("DID is not registered or was recently deleted. DID has to be registered first")

        new_private_key = PrivateKey.from_string(new_private_key_der)
        new_key_type = get_key_type(new_private_key)

        topic_update_options = HcsTopicOptions(
            admin_key=new_private_key.public_key(), submit_key=new_private_key.public_key()
        )
        await self._hcs_topic_service.update_topic(
            cast(str, self.topic_id), topic_update_options, [cast(PrivateKey, self._private_key), new_private_key]
        )

        self._private_key = new_private_key
        self._key_type = new_key_type

        hcs_event = HcsDidUpdateDidOwnerEvent(
            id_=f"{self.identifier}#did-root-key",
            controller=controller,
            public_key=self._private_key.public_key(),
            type_=self._key_type,
        )

        await self._submit_transaction(DidDocumentOperation.UPDATE, hcs_event)

    async def resolve(self) -> DidDocument:
        """
        Resolve DID document for registered instance.

        Returns:
            object: DID document
        """
        if not self.topic_id or not self.identifier:
            raise DidException("DID is not registered")

        result = await HcsMessageResolver(self.topic_id, HcsDidMessageEnvelope).execute(self._client)
        await self._handle_resolution_result(cast(list[HcsDidMessageEnvelope], result))

        return cast(DidDocument, self.document)

    async def delete(self):
        """Delete (deactivate) registered DID instance."""
        self._assert_can_submit_transaction()

        await self._submit_transaction(DidDocumentOperation.DELETE, HcsDidDeleteEvent())

    async def add_service(self, id_: str, service_type: DidServiceType, service_endpoint: str):
        """Add Service to DID document

        Args:
            id_: Service ID to create
            service_type: DID service type
            service_endpoint: Service endpoint
        """
        await self._add_or_update_service(
            DidDocumentOperation.CREATE, id_=id_, type_=service_type, service_endpoint=service_endpoint
        )

    async def update_service(self, id_: str, service_type: DidServiceType, service_endpoint: str):
        """Update existing DID document service

        Args:
            id_: Service ID to update
            service_type: DID service type
            service_endpoint: Service endpoint
        """
        await self._add_or_update_service(
            DidDocumentOperation.UPDATE, id_=id_, type_=service_type, service_endpoint=service_endpoint
        )

    async def revoke_service(self, id_: str):
        """Revoke existing DID document service

        Args:
            id_: Service ID to revoke
        """
        self._assert_can_submit_transaction()

        hcs_event = HcsDidRevokeServiceEvent(id_)
        await self._submit_transaction(DidDocumentOperation.REVOKE, hcs_event)

    async def add_verification_method(
        self,
        id_: str,
        controller: str,
        public_key_der: str,
        type_: SupportedKeyType,
    ):
        """Add verification method to DID document

        Args:
            id_: Verification method ID to create
            controller: Verification method controller ID
            public_key_der: Verification method public key encoded in DER format
            type_: Verification method key type
        """
        await self._add_or_update_verification_method(
            DidDocumentOperation.CREATE,
            id_=id_,
            controller=controller,
            public_key=PublicKey.from_string(public_key_der),
            type_=type_,
        )

    async def update_verification_method(
        self,
        id_: str,
        controller: str,
        public_key_der: str,
        type_: SupportedKeyType,
    ):
        """Update existing DID document verification method

        Args:
            id_: Verification method ID to update
            controller: Verification method controller ID
            public_key_der: Verification method public key encoded in DER format
            type_: Verification method key type
        """
        await self._add_or_update_verification_method(
            DidDocumentOperation.UPDATE,
            id_=id_,
            controller=controller,
            public_key=PublicKey.from_string(public_key_der),
            type_=type_,
        )

    async def revoke_verification_method(self, id_: str):
        """Revoke existing DID document verification method

        Args:
            id_: Verification method ID to revoke
        """
        self._assert_can_submit_transaction()

        hcs_event = HcsDidRevokeVerificationMethodEvent(id_)
        await self._submit_transaction(DidDocumentOperation.REVOKE, hcs_event)

    async def add_verification_relationship(
        self,
        id_: str,
        controller: str,
        public_key_der: str,
        relationship_type: VerificationRelationshipType,
        type_: SupportedKeyType,
    ):
        """Add verification relationship to DID document

        Args:
            id_: Verification relationship ID to create
            controller: Verification relationship controller ID
            public_key_der: Verification relationship public key encoded in DER format
            relationship_type: Verification relationship type
            type_: Verification relationship key type
        """
        await self._add_or_update_verification_relationship(
            DidDocumentOperation.CREATE,
            id_=id_,
            controller=controller,
            public_key=PublicKey.from_string(public_key_der),
            relationship_type=relationship_type,
            type_=type_,
        )

    async def update_verification_relationship(
        self,
        id_: str,
        controller: str,
        public_key_der: str,
        relationship_type: VerificationRelationshipType,
        type_: SupportedKeyType,
    ):
        """Update existing DID document verification relationship

        Args:
            id_: Verification relationship ID to update
            controller: Verification relationship controller ID
            public_key_der: Verification relationship public key encoded in DER format
            relationship_type: Verification relationship type
            type_: Verification relationship key type
        """
        await self._add_or_update_verification_relationship(
            DidDocumentOperation.UPDATE,
            id_=id_,
            public_key=PublicKey.from_string(public_key_der),
            controller=controller,
            relationship_type=relationship_type,
            type_=type_,
        )

    async def revoke_verification_relationship(self, id_: str, relationship_type: VerificationRelationshipType):
        """Revoke existing DID document verification relationship

        Args:
            id_: Verification relationship ID to revoke
            relationship_type: Verification relationship type
        """
        self._assert_can_submit_transaction()

        hcs_event = HcsDidRevokeVerificationRelationshipEvent(id_, relationship_type)
        await self._submit_transaction(DidDocumentOperation.REVOKE, hcs_event)

    async def _submit_transaction(self, operation: DidDocumentOperation, event: HcsDidEvent):
        if not self.topic_id or not self.identifier or not self._private_key:
            raise Exception("Cannot submit transaction: topic_id, identifier and private_key must be set")

        message = HcsDidMessage(operation, self.identifier, event)
        envelope = HcsDidMessageEnvelope(message)
        envelope.sign(self._private_key)

        def build_did_transaction(message_submit_transaction: TopicMessageSubmitTransaction) -> Transaction:
            message_submit_transaction.transaction_fee = MAX_TRANSACTION_FEE.to_tinybars()  # pyright: ignore [reportAttributeAccessIssue]
            return message_submit_transaction.freeze_with(self._client).sign(self._private_key)

        await HcsMessageTransaction(self.topic_id, envelope, build_did_transaction).execute(self._client)

    async def _add_or_update_service(
        self, operation: Literal[DidDocumentOperation.CREATE, DidDocumentOperation.UPDATE], **kwargs
    ):
        self._assert_can_submit_transaction()

        await self._submit_transaction(operation, HcsDidUpdateServiceEvent(**kwargs))

    async def _add_or_update_verification_method(
        self, operation: Literal[DidDocumentOperation.CREATE, DidDocumentOperation.UPDATE], **kwargs
    ):
        self._assert_can_submit_transaction()

        await self._submit_transaction(operation, HcsDidUpdateVerificationMethodEvent(**kwargs))

    async def _add_or_update_verification_relationship(
        self, operation: Literal[DidDocumentOperation.CREATE, DidDocumentOperation.UPDATE], **kwargs
    ):
        self._assert_can_submit_transaction()

        await self._submit_transaction(operation, HcsDidUpdateVerificationRelationshipEvent(**kwargs))

    async def _handle_resolution_result(self, result: list[HcsDidMessageEnvelope]):
        if not self.identifier:
            raise Exception("Cannot handle DID resolution result: DID identifier is not defined")

        self._messages = [cast(HcsDidMessage, envelope.message) for envelope in result]
        self.document = DidDocument(self.identifier)
        await self.document.process_messages(self._messages)

    def _assert_can_submit_transaction(self):
        if not self.identifier:
            raise DidException("DID is not registered")

        if not self._private_key or not self._key_type:
            raise DidException("Private key is required to submit DID event transaction")

add_service(id_, service_type, service_endpoint) async

Add Service to DID document

Parameters:

Name	Type	Description	Default
id_	str	Service ID to create	required
service_type	DidServiceType	DID service type	required
service_endpoint	str	Service endpoint	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def add_service(self, id_: str, service_type: DidServiceType, service_endpoint: str):
    """Add Service to DID document

    Args:
        id_: Service ID to create
        service_type: DID service type
        service_endpoint: Service endpoint
    """
    await self._add_or_update_service(
        DidDocumentOperation.CREATE, id_=id_, type_=service_type, service_endpoint=service_endpoint
    )

add_verification_method(id_, controller, public_key_der, type_) async

Add verification method to DID document

Parameters:

Name	Type	Description	Default
id_	str	Verification method ID to create	required
controller	str	Verification method controller ID	required
public_key_der	str	Verification method public key encoded in DER format	required
type_	SupportedKeyType	Verification method key type	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def add_verification_method(
    self,
    id_: str,
    controller: str,
    public_key_der: str,
    type_: SupportedKeyType,
):
    """Add verification method to DID document

    Args:
        id_: Verification method ID to create
        controller: Verification method controller ID
        public_key_der: Verification method public key encoded in DER format
        type_: Verification method key type
    """
    await self._add_or_update_verification_method(
        DidDocumentOperation.CREATE,
        id_=id_,
        controller=controller,
        public_key=PublicKey.from_string(public_key_der),
        type_=type_,
    )

add_verification_relationship(id_, controller, public_key_der, relationship_type, type_) async

Add verification relationship to DID document

Parameters:

Name	Type	Description	Default
id_	str	Verification relationship ID to create	required
controller	str	Verification relationship controller ID	required
public_key_der	str	Verification relationship public key encoded in DER format	required
relationship_type	VerificationRelationshipType	Verification relationship type	required
type_	SupportedKeyType	Verification relationship key type	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def add_verification_relationship(
    self,
    id_: str,
    controller: str,
    public_key_der: str,
    relationship_type: VerificationRelationshipType,
    type_: SupportedKeyType,
):
    """Add verification relationship to DID document

    Args:
        id_: Verification relationship ID to create
        controller: Verification relationship controller ID
        public_key_der: Verification relationship public key encoded in DER format
        relationship_type: Verification relationship type
        type_: Verification relationship key type
    """
    await self._add_or_update_verification_relationship(
        DidDocumentOperation.CREATE,
        id_=id_,
        controller=controller,
        public_key=PublicKey.from_string(public_key_der),
        relationship_type=relationship_type,
        type_=type_,
    )

change_owner(controller, new_private_key_der) async

Change DID Owner (controller).

Parameters:

Name	Type	Description	Default
controller	str	Identifier of new DID Owner	required
new_private_key_der	str	New DID Owner private key encoded in DER format	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def change_owner(self, controller: str, new_private_key_der: str):
    """
    Change DID Owner (controller).

    Args:
        controller: Identifier of new DID Owner
        new_private_key_der: New DID Owner private key encoded in DER format
    """
    self._assert_can_submit_transaction()

    document = await self.resolve()
    if not document.controller:
        raise DidException("DID is not registered or was recently deleted. DID has to be registered first")

    new_private_key = PrivateKey.from_string(new_private_key_der)
    new_key_type = get_key_type(new_private_key)

    topic_update_options = HcsTopicOptions(
        admin_key=new_private_key.public_key(), submit_key=new_private_key.public_key()
    )
    await self._hcs_topic_service.update_topic(
        cast(str, self.topic_id), topic_update_options, [cast(PrivateKey, self._private_key), new_private_key]
    )

    self._private_key = new_private_key
    self._key_type = new_key_type

    hcs_event = HcsDidUpdateDidOwnerEvent(
        id_=f"{self.identifier}#did-root-key",
        controller=controller,
        public_key=self._private_key.public_key(),
        type_=self._key_type,
    )

    await self._submit_transaction(DidDocumentOperation.UPDATE, hcs_event)

delete() async

Delete (deactivate) registered DID instance.

Source code in hiero_did_sdk_python/did/hedera_did.py

async def delete(self):
    """Delete (deactivate) registered DID instance."""
    self._assert_can_submit_transaction()

    await self._submit_transaction(DidDocumentOperation.DELETE, HcsDidDeleteEvent())

register() async

Register (create) DID instance in Hedera network

Source code in hiero_did_sdk_python/did/hedera_did.py

async def register(self):
    """Register (create) DID instance in Hedera network"""
    if not self._private_key or not self._key_type:
        raise DidException("Private key is required to register new DID")

    if self.identifier:
        document = await self.resolve()
        if document.controller:
            raise DidException("DID is already registered")
    else:
        topic_options = HcsTopicOptions(
            admin_key=self._private_key.public_key(), submit_key=self._private_key.public_key()
        )

        self.topic_id = await self._hcs_topic_service.create_topic(topic_options, [self._private_key])

        self.network = self._client.network.network
        self.identifier = build_identifier(
            self.network,
            multibase_encode(bytes(self._private_key.public_key().to_bytes_raw()), "base58btc"),
            self.topic_id,
        )

    hcs_event = HcsDidUpdateDidOwnerEvent(
        id_=f"{self.identifier}#did-root-key",
        controller=self.identifier,
        public_key=self._private_key.public_key(),
        type_=self._key_type,
    )

    await self._submit_transaction(DidDocumentOperation.CREATE, hcs_event)

resolve() async

Resolve DID document for registered instance.

Returns:

Name	Type	Description
object	DidDocument	DID document

Source code in hiero_did_sdk_python/did/hedera_did.py

async def resolve(self) -> DidDocument:
    """
    Resolve DID document for registered instance.

    Returns:
        object: DID document
    """
    if not self.topic_id or not self.identifier:
        raise DidException("DID is not registered")

    result = await HcsMessageResolver(self.topic_id, HcsDidMessageEnvelope).execute(self._client)
    await self._handle_resolution_result(cast(list[HcsDidMessageEnvelope], result))

    return cast(DidDocument, self.document)

revoke_service(id_) async

Revoke existing DID document service

Parameters:

Name	Type	Description	Default
id_	str	Service ID to revoke	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def revoke_service(self, id_: str):
    """Revoke existing DID document service

    Args:
        id_: Service ID to revoke
    """
    self._assert_can_submit_transaction()

    hcs_event = HcsDidRevokeServiceEvent(id_)
    await self._submit_transaction(DidDocumentOperation.REVOKE, hcs_event)

revoke_verification_method(id_) async

Revoke existing DID document verification method

Parameters:

Name	Type	Description	Default
id_	str	Verification method ID to revoke	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def revoke_verification_method(self, id_: str):
    """Revoke existing DID document verification method

    Args:
        id_: Verification method ID to revoke
    """
    self._assert_can_submit_transaction()

    hcs_event = HcsDidRevokeVerificationMethodEvent(id_)
    await self._submit_transaction(DidDocumentOperation.REVOKE, hcs_event)

revoke_verification_relationship(id_, relationship_type) async

Revoke existing DID document verification relationship

Parameters:

Name	Type	Description	Default
id_	str	Verification relationship ID to revoke	required
relationship_type	VerificationRelationshipType	Verification relationship type	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def revoke_verification_relationship(self, id_: str, relationship_type: VerificationRelationshipType):
    """Revoke existing DID document verification relationship

    Args:
        id_: Verification relationship ID to revoke
        relationship_type: Verification relationship type
    """
    self._assert_can_submit_transaction()

    hcs_event = HcsDidRevokeVerificationRelationshipEvent(id_, relationship_type)
    await self._submit_transaction(DidDocumentOperation.REVOKE, hcs_event)

update_service(id_, service_type, service_endpoint) async

Update existing DID document service

Parameters:

Name	Type	Description	Default
id_	str	Service ID to update	required
service_type	DidServiceType	DID service type	required
service_endpoint	str	Service endpoint	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def update_service(self, id_: str, service_type: DidServiceType, service_endpoint: str):
    """Update existing DID document service

    Args:
        id_: Service ID to update
        service_type: DID service type
        service_endpoint: Service endpoint
    """
    await self._add_or_update_service(
        DidDocumentOperation.UPDATE, id_=id_, type_=service_type, service_endpoint=service_endpoint
    )

update_verification_method(id_, controller, public_key_der, type_) async

Update existing DID document verification method

Parameters:

Name	Type	Description	Default
id_	str	Verification method ID to update	required
controller	str	Verification method controller ID	required
public_key_der	str	Verification method public key encoded in DER format	required
type_	SupportedKeyType	Verification method key type	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def update_verification_method(
    self,
    id_: str,
    controller: str,
    public_key_der: str,
    type_: SupportedKeyType,
):
    """Update existing DID document verification method

    Args:
        id_: Verification method ID to update
        controller: Verification method controller ID
        public_key_der: Verification method public key encoded in DER format
        type_: Verification method key type
    """
    await self._add_or_update_verification_method(
        DidDocumentOperation.UPDATE,
        id_=id_,
        controller=controller,
        public_key=PublicKey.from_string(public_key_der),
        type_=type_,
    )

update_verification_relationship(id_, controller, public_key_der, relationship_type, type_) async

Update existing DID document verification relationship

Parameters:

Name	Type	Description	Default
id_	str	Verification relationship ID to update	required
controller	str	Verification relationship controller ID	required
public_key_der	str	Verification relationship public key encoded in DER format	required
relationship_type	VerificationRelationshipType	Verification relationship type	required
type_	SupportedKeyType	Verification relationship key type	required

Source code in hiero_did_sdk_python/did/hedera_did.py

async def update_verification_relationship(
    self,
    id_: str,
    controller: str,
    public_key_der: str,
    relationship_type: VerificationRelationshipType,
    type_: SupportedKeyType,
):
    """Update existing DID document verification relationship

    Args:
        id_: Verification relationship ID to update
        controller: Verification relationship controller ID
        public_key_der: Verification relationship public key encoded in DER format
        relationship_type: Verification relationship type
        type_: Verification relationship key type
    """
    await self._add_or_update_verification_relationship(
        DidDocumentOperation.UPDATE,
        id_=id_,
        public_key=PublicKey.from_string(public_key_der),
        controller=controller,
        relationship_type=relationship_type,
        type_=type_,
    )

Models and types

DidDocument

Bases: Serializable

DID document representation

Attributes:

Name	Type	Description
created	float | None	Creation timestamp
updated	float | None	Last update timestamp
version_id	str | None	DID document version ID (equals to last update timestamp)
deactivated	bool	DID document deactivation status
controller	dict | None	Dictionary representing DID document controller info
services	dict	DID document services dictionary
verification_methods	dict	DID document verification methods dictionary
verification_methods	dict	DID document verification relationships dictionary

Source code in hiero_did_sdk_python/did/did_document.py

class DidDocument(Serializable):
    """DID document representation

    Attributes:
        created: Creation timestamp
        updated: Last update timestamp
        version_id: DID document version ID (equals to last update timestamp)
        deactivated: DID document deactivation status
        controller: Dictionary representing DID document controller info
        services: DID document services dictionary
        verification_methods: DID document verification methods dictionary
        verification_methods: DID document verification relationships dictionary

    """

    def __init__(self, id_: str):
        self.id_ = id_
        self.context = DID_DOCUMENT_CONTEXT

        self.created: float | None = None
        self.updated: float | None = None
        self.version_id: str | None = None
        self.deactivated: bool = False

        self.controller: dict | None = None
        self.services: dict = {}
        self.verification_methods: dict = {}

        self.verification_relationships: dict = {
            DidDocumentJsonProperties.AUTHENTICATION.value: [],
            DidDocumentJsonProperties.ASSERTION_METHOD.value: [],
            DidDocumentJsonProperties.KEY_AGREEMENT.value: [],
            DidDocumentJsonProperties.CAPABILITY_INVOCATION.value: [],
            DidDocumentJsonProperties.CAPABILITY_DELEGATION.value: [],
        }

    async def process_messages(self, messages: list[HcsDidMessage]):
        """
        Process HCS DID messages - apply DID document state changes according to events.

        Args:
            messages: HCS DID messages to process

        """
        for message in messages:
            if not self.controller and message.operation == DidDocumentOperation.CREATE:
                event_target = message.event.event_target
                if event_target != HcsDidEventTarget.DID_OWNER and event_target != HcsDidEventTarget.DID_DOCUMENT:
                    LOGGER.warning("DID document is not registered, skipping DID update event...")
                    continue

            match message.operation:
                case DidDocumentOperation.CREATE:
                    await self._process_create_message(message)
                case DidDocumentOperation.UPDATE:
                    self._process_update_message(message)
                case DidDocumentOperation.DELETE:
                    self._process_delete_message(message)
                case DidDocumentOperation.REVOKE:
                    self._process_revoke_message(message)
                case _:
                    LOGGER.warning(f"Operation {message.operation} is not supported, skipping DID event...")

    @classmethod
    def from_json_payload(cls, payload: dict):
        raise Exception("DidDocument deserialization is not implemented")

    def get_json_payload(self):
        root_object: dict = {
            DidDocumentJsonProperties.CONTEXT.value: self.context,
            DidDocumentJsonProperties.ID.value: self.id_,
        }

        if self.controller and self.id_ != self.controller.get("controller"):
            root_object[DidDocumentJsonProperties.CONTROLLER.value] = (
                self.controller.get("controller") or self.controller
            )

        root_object[DidDocumentJsonProperties.VERIFICATION_METHOD.value] = list(self.verification_methods.values())

        root_object[DidDocumentJsonProperties.ASSERTION_METHOD.value] = [
            *self.verification_relationships[DidDocumentJsonProperties.ASSERTION_METHOD.value],
        ]

        root_object[DidDocumentJsonProperties.AUTHENTICATION.value] = [
            *self.verification_relationships[DidDocumentJsonProperties.AUTHENTICATION.value],
        ]

        if self.controller:
            controller_id = self.controller.get("id")

            root_object[DidDocumentJsonProperties.VERIFICATION_METHOD.value].insert(0, self.controller)
            root_object[DidDocumentJsonProperties.ASSERTION_METHOD.value].insert(0, controller_id)
            root_object[DidDocumentJsonProperties.AUTHENTICATION.value].insert(0, controller_id)

        if len(self.verification_relationships[DidDocumentJsonProperties.KEY_AGREEMENT.value]) > 0:
            root_object[DidDocumentJsonProperties.KEY_AGREEMENT.value] = [
                *self.verification_relationships[DidDocumentJsonProperties.KEY_AGREEMENT.value],
            ]
        if len(self.verification_relationships[DidDocumentJsonProperties.CAPABILITY_INVOCATION.value]) > 0:
            root_object[DidDocumentJsonProperties.CAPABILITY_INVOCATION.value] = [
                *self.verification_relationships[DidDocumentJsonProperties.CAPABILITY_INVOCATION.value],
            ]
        if len(self.verification_relationships[DidDocumentJsonProperties.CAPABILITY_DELEGATION.value]) > 0:
            root_object[DidDocumentJsonProperties.CAPABILITY_DELEGATION.value] = [
                *self.verification_relationships[DidDocumentJsonProperties.CAPABILITY_DELEGATION.value],
            ]

        if len(self.services) > 0:
            root_object[DidDocumentJsonProperties.SERVICE.value] = list(self.services.values())

        return root_object

    async def _process_create_message(self, message: HcsDidMessage):  # noqa: C901
        event = message.event

        match event.event_target:
            case HcsDidEventTarget.DID_DOCUMENT:
                document = await download_ipfs_document_by_cid(cast(HcsDidCreateDidDocumentEvent, event).cid)

                if document[DidDocumentJsonProperties.ID] != self.id_:
                    raise ValueError("Document ID does not match did")

                self.controller = document[DidDocumentJsonProperties.CONTROLLER]

                self.services = {
                    service["id"]: service for service in document.get(DidDocumentJsonProperties.SERVICE, [])
                }

                self.verification_methods = {
                    verificationMethod["id"]: verificationMethod
                    for verificationMethod in document.get(DidDocumentJsonProperties.VERIFICATION_METHOD, [])
                }

                self.verification_relationships[DidDocumentJsonProperties.ASSERTION_METHOD] = document.get(
                    DidDocumentJsonProperties.ASSERTION_METHOD, []
                )
                self.verification_relationships[DidDocumentJsonProperties.AUTHENTICATION] = document.get(
                    DidDocumentJsonProperties.AUTHENTICATION, []
                )
                self.verification_relationships[DidDocumentJsonProperties.KEY_AGREEMENT] = document.get(
                    DidDocumentJsonProperties.KEY_AGREEMENT, []
                )
                self.verification_relationships[DidDocumentJsonProperties.CAPABILITY_INVOCATION] = document.get(
                    DidDocumentJsonProperties.CAPABILITY_INVOCATION, []
                )
                self.verification_relationships[DidDocumentJsonProperties.CAPABILITY_DELEGATION] = document.get(
                    DidDocumentJsonProperties.CAPABILITY_DELEGATION, []
                )
            case HcsDidEventTarget.DID_OWNER:
                if self.controller:
                    LOGGER.warning(f"DID owner is already registered: {self.controller}, skipping event...")
                    return

                self.controller = cast(HcsDidUpdateDidOwnerEvent, event).get_owner_def()
                self._on_activated(message.timestamp)
            case HcsDidEventTarget.SERVICE:
                update_service_event = cast(HcsDidUpdateServiceEvent, event)
                event_id = update_service_event.id_

                if event_id in self.services:
                    LOGGER.warning(f"Duplicate create Service event ID: {event_id}, skipping event...")
                    return

                self.services[event_id] = update_service_event.get_service_def()
                self._on_updated(message.timestamp)
            case HcsDidEventTarget.VERIFICATION_METHOD:
                update_verification_method_event = cast(HcsDidUpdateVerificationMethodEvent, event)
                event_id = update_verification_method_event.id_

                if event_id in self.verification_methods:
                    LOGGER.warning(f"Duplicate create Verification Method event ID: {event_id}, skipping event...")
                    return

                self.verification_methods[event_id] = update_verification_method_event.get_verification_method_def()
                self._on_updated(message.timestamp)
            case HcsDidEventTarget.VERIFICATION_RELATIONSHIP:
                update_verification_relationship_event = cast(HcsDidUpdateVerificationRelationshipEvent, event)
                relationship_type = update_verification_relationship_event.relationship_type
                event_id = update_verification_relationship_event.id_

                if relationship_type not in self.verification_relationships:
                    LOGGER.warning(
                        f"Create verification Relationship event with type {relationship_type} is not supported, skipping event..."
                    )
                    return

                if event_id in self.verification_relationships[relationship_type]:
                    LOGGER.warning(
                        f"Duplicate create Verification Relationship event ID: {event_id}, skipping event..."
                    )
                    return

                self.verification_relationships[relationship_type].append(event_id)

                if event_id not in self.verification_methods:
                    self.verification_methods[event_id] = (
                        update_verification_relationship_event.get_verification_method_def()
                    )
                self._on_updated(message.timestamp)
            case _:
                LOGGER.warning(f"Create {event.event_target} operation is not supported, skipping event...")

    def _process_update_message(self, message: HcsDidMessage):
        event = message.event

        match event.event_target:
            case HcsDidEventTarget.DID_OWNER:
                self.controller = cast(HcsDidUpdateDidOwnerEvent, event).get_owner_def()
                self._on_updated(message.timestamp)
            case HcsDidEventTarget.SERVICE:
                update_service_event = cast(HcsDidUpdateServiceEvent, event)
                event_id = update_service_event.id_

                if event_id not in self.services:
                    LOGGER.warning(f"Service with ID: {event_id} is not found on the document, skipping event...")
                    return

                self.services[event_id] = update_service_event.get_service_def()
                self._on_updated(message.timestamp)
            case HcsDidEventTarget.VERIFICATION_METHOD:
                update_verification_method_event = cast(HcsDidUpdateVerificationMethodEvent, event)
                event_id = update_verification_method_event.id_

                if event_id not in self.verification_methods:
                    LOGGER.warning(
                        f"Verification Method with ID: {event_id} is not found on the document, skipping event..."
                    )
                    return

                self.verification_methods[event_id] = update_verification_method_event.get_verification_method_def()
                self._on_updated(message.timestamp)
            case HcsDidEventTarget.VERIFICATION_RELATIONSHIP:
                update_verification_relationship_event = cast(HcsDidUpdateVerificationRelationshipEvent, event)
                relationship_type = update_verification_relationship_event.relationship_type
                event_id = update_verification_relationship_event.id_

                if relationship_type not in self.verification_relationships:
                    LOGGER.warning(
                        f"Update verification Relationship event with type {relationship_type} is not supported, skipping event..."
                    )
                    return

                if event_id not in self.verification_relationships[relationship_type]:
                    LOGGER.warning(
                        f"Verification Relationship with ID: {event_id} is not found on the document, skipping event..."
                    )
                    return

                self.verification_methods[event_id] = (
                    update_verification_relationship_event.get_verification_method_def()
                )
                self._on_updated(message.timestamp)
            case _:
                LOGGER.warning(f"Update {event.event_target} operation is not supported, skipping event...")

    def _process_revoke_message(self, message: HcsDidMessage):
        event = message.event

        match event.event_target:
            case HcsDidEventTarget.SERVICE:
                revoke_service_event = cast(HcsDidRevokeServiceEvent, event)
                event_id = revoke_service_event.id_

                if event_id not in self.services:
                    LOGGER.warning(f"Service with ID: {event_id} is not found on the document, skipping event...")
                    return

                del self.services[event_id]
                self._on_updated(message.timestamp)
            case HcsDidEventTarget.VERIFICATION_METHOD:
                revoke_verification_method_event = cast(HcsDidRevokeVerificationMethodEvent, event)
                event_id = revoke_verification_method_event.id_

                if event_id not in self.verification_methods:
                    LOGGER.warning(
                        f"Verification Method with ID: {event_id} is not found on the document, skipping event..."
                    )
                    return

                del self.verification_methods[event_id]

                for type_key in self.verification_relationships:
                    self.verification_relationships[type_key] = list(
                        filter(lambda id_: id_ != event_id, self.verification_relationships[type_key])
                    )

                self._on_updated(message.timestamp)
            case HcsDidEventTarget.VERIFICATION_RELATIONSHIP:
                revoke_verification_relationship_event = cast(HcsDidUpdateVerificationRelationshipEvent, event)
                relationship_type = revoke_verification_relationship_event.relationship_type
                event_id = revoke_verification_relationship_event.id_

                if relationship_type not in self.verification_relationships:
                    LOGGER.warning(
                        f"Revoke verification Relationship event with type {relationship_type} is not supported, skipping event..."
                    )
                    return

                if event_id not in self.verification_relationships[relationship_type]:
                    LOGGER.warning(
                        f"Verification Relationship with ID: {event_id} is not found on the document, skipping event..."
                    )
                    return

                self.verification_relationships[relationship_type] = list(
                    filter(lambda id_: id_ != event_id, self.verification_relationships[relationship_type])
                )

                can_delete_verification_method = all(
                    event_id not in rel for rel in self.verification_relationships.values()
                )

                if can_delete_verification_method:
                    del self.verification_methods[event_id]

                self._on_updated(message.timestamp)
            case _:
                LOGGER.warning(f"Revoke {event.event_target} operation is not supported, skipping event...")

    def _process_delete_message(self, message: HcsDidMessage):
        event = message.event

        match event.event_target:
            case HcsDidEventTarget.DOCUMENT:
                self.controller = None
                self.services.clear()
                self.verification_methods.clear()
                for type_key in self.verification_relationships:
                    self.verification_relationships[type_key] = []
                self._on_deactivated()
            case _:
                LOGGER.warning(f"Delete {event.event_target} operation is not supported, skipping event...")

    def _on_activated(self, timestamp: float):
        self.created = timestamp
        self.updated = timestamp
        self.deactivated = False
        self.version_id = str(timestamp)

    def _on_updated(self, timestamp: float):
        self.updated = timestamp
        self.version_id = str(timestamp)

    def _on_deactivated(self):
        self.created = None
        self.updated = None
        self.deactivated = True
        self.version_id = None

process_messages(messages) async

Process HCS DID messages - apply DID document state changes according to events.

Parameters:

Name	Type	Description	Default
messages	list[HcsDidMessage]	HCS DID messages to process	required

Source code in hiero_did_sdk_python/did/did_document.py

async def process_messages(self, messages: list[HcsDidMessage]):
    """
    Process HCS DID messages - apply DID document state changes according to events.

    Args:
        messages: HCS DID messages to process

    """
    for message in messages:
        if not self.controller and message.operation == DidDocumentOperation.CREATE:
            event_target = message.event.event_target
            if event_target != HcsDidEventTarget.DID_OWNER and event_target != HcsDidEventTarget.DID_DOCUMENT:
                LOGGER.warning("DID document is not registered, skipping DID update event...")
                continue

        match message.operation:
            case DidDocumentOperation.CREATE:
                await self._process_create_message(message)
            case DidDocumentOperation.UPDATE:
                self._process_update_message(message)
            case DidDocumentOperation.DELETE:
                self._process_delete_message(message)
            case DidDocumentOperation.REVOKE:
                self._process_revoke_message(message)
            case _:
                LOGGER.warning(f"Operation {message.operation} is not supported, skipping DID event...")

DidErrorCode

Bases: StrEnum

Enum for DID-related error codes

Source code in hiero_did_sdk_python/did/did_error.py

class DidErrorCode(StrEnum):
    """Enum for DID-related error codes"""

    """Generic (unknown) error"""
    GENERIC = "generic"

    """Invalid DID identifier string. Can be caused by non-Hedera DID method prefix, invalid or missing Topic ID and generally incorrect format"""
    INVALID_DID_STRING = "invalid_did_string"

    """Invalid Hedera network specified in DID identifier"""
    INVALID_NETWORK = "invalid_network"

    """Specified DID is not found"""
    DID_NOT_FOUND = "did_not_found"

GENERIC = 'generic' class-attribute instance-attribute

Invalid DID identifier string. Can be caused by non-Hedera DID method prefix, invalid or missing Topic ID and generally incorrect format

INVALID_DID_STRING = 'invalid_did_string' class-attribute instance-attribute

Invalid Hedera network specified in DID identifier

INVALID_NETWORK = 'invalid_network' class-attribute instance-attribute

Specified DID is not found

DidException

Bases: Exception

Class for DID-related exceptions

Parameters:

Name	Type	Description	Default
message	str	Error message	required
code	DidErrorCode	DID Error code	GENERIC

Source code in hiero_did_sdk_python/did/did_error.py

class DidException(Exception):
    """Class for DID-related exceptions

    Args:
        message: Error message
        code: DID Error code
    """

    def __init__(self, message: str, code: DidErrorCode = DidErrorCode.GENERIC):
        super().__init__(message)

        self.code = code