Source code for aries_cloudagent.protocols.didcomm_prefix

"""DIDComm prefix management."""
import re

from enum import Enum
from os import environ
from typing import Mapping

QUALIFIED = re.compile(r"^[a-zA-Z\-\+]+:.+")

[docs]def qualify(msg_type: str, prefix: str): """Qualify a message type with a prefix, if unqualified.""" return msg_type if QUALIFIED.match(msg_type or "") else f"{prefix}/{msg_type}"
[docs]class DIDCommPrefix(Enum): """Enum for DIDComm Prefix, old or new style, per Aries RFC 384.""" NEW = "" OLD = "did:sov:BzCbsNYhMrjHiqZDTUASHg;spec"
[docs] @staticmethod def set(settings: Mapping): """Set current DIDComm prefix value in environment.""" environ["DIDCOMM_PREFIX"] = ( DIDCommPrefix.NEW.value if settings.get("emit_new_didcomm_prefix") else DIDCommPrefix.OLD.value )
[docs] def qualify(self, msg_type: str = None) -> str: """Qualify input message type with prefix and separator.""" return qualify(msg_type, self.value)
[docs] @classmethod def qualify_all(cls, messages: dict) -> dict: """Apply all known prefixes to a dictionary of message types.""" return {qualify(k, pfx.value): v for pfx in cls for k, v in messages.items()}
[docs] @staticmethod def qualify_current(slug: str = None) -> str: """Qualify input slug with prefix currently in effect and separator.""" return qualify(slug, environ.get("DIDCOMM_PREFIX", DIDCommPrefix.OLD.value))
[docs] @staticmethod def unqualify(qual: str) -> str: """Strip prefix and separator from input, if present, and return result.""" for pfx in DIDCommPrefix: if (qual or "").startswith(f"{pfx.value}/"): return qual.split(f"{pfx.value}/")[1] return qual