acapy_agent.utils package

Subpackages

• acapy_agent.utils.multiformats package
  • Submodules
  • acapy_agent.utils.multiformats.multibase module
    • Base58BtcEncoder
      • Base58BtcEncoder.character
      • Base58BtcEncoder.decode()
      • Base58BtcEncoder.encode()
      • Base58BtcEncoder.name
    • Encoding
      • Encoding.base58btc
      • Encoding.from_character()
      • Encoding.from_name()
    • MultibaseEncoder
      • MultibaseEncoder.character
      • MultibaseEncoder.decode()
      • MultibaseEncoder.encode()
      • MultibaseEncoder.name
    • decode()
    • encode()
  • acapy_agent.utils.multiformats.multicodec module
    • Multicodec
      • Multicodec.code
      • Multicodec.name
    • SupportedCodecs
      • SupportedCodecs.bls12381g1
      • SupportedCodecs.bls12381g1g2
      • SupportedCodecs.bls12381g2
      • SupportedCodecs.by_name()
      • SupportedCodecs.ed25519_pub
      • SupportedCodecs.for_data()
      • SupportedCodecs.p256_pub
      • SupportedCodecs.secp256k1_pub
      • SupportedCodecs.x25519_pub
    • multicodec()
    • unwrap()
    • wrap()

Submodules

acapy_agent.utils.classloader module

The classloader provides utilities to dynamically load classes and modules.

class acapy_agent.utils.classloader.ClassLoader

Bases: object

Class used to load classes from modules dynamically.

classmethod load_class(class_name: str, default_module: str | None = None, package: str | None = None)

Resolve a complete class path (ie. typing.Dict) to the class itself.

Parameters:

• class_name – the class name

• default_module – the default module to load, if not part of in the class name

• package – the parent package to search for the module

Returns:

The resolved class

Raises:

• ClassNotFoundError – If the class could not be resolved at path

• ModuleLoadError – If there was an error loading the module

classmethod load_module(mod_path: str, package: str | None = None) → ModuleType | None

Load a module by its absolute path.

Parameters:

• mod_path – the absolute or relative module path

• package – the parent package to search for the module

Returns:

The resolved module or None if the module cannot be found

Raises:

ModuleLoadError – If there was an error loading the module

classmethod load_subclass_of(base_class: Type, mod_path: str, package: str | None = None)

Resolve an implementation of a base path within a module.

Parameters:

• base_class – the base class being implemented

• mod_path – the absolute module path

• package – the parent package to search for the module

Returns:

The resolved class

Raises:

• ClassNotFoundError – If the module or class implementation could not be found

• ModuleLoadError – If there was an error loading the module

classmethod scan_subpackages(package: str) → Sequence[str]

Return a list of sub-packages defined under a named package.

exception acapy_agent.utils.classloader.ClassNotFoundError(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Class not found error.

class acapy_agent.utils.classloader.DeferLoad(cls_path: str)

Bases: object

Helper to defer loading of a class definition.

property resolved

Accessor for the resolved class instance.

exception acapy_agent.utils.classloader.ModuleLoadError(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Module load error.

acapy_agent.utils.dependencies module

Dependency related util methods.

acapy_agent.utils.dependencies.assert_ursa_bbs_signatures_installed()

Assert ursa_bbs_signatures module is installed.

acapy_agent.utils.dependencies.is_ursa_bbs_signatures_module_installed()

Check whether ursa_bbs_signatures module is installed.

Returns:

Whether ursa_bbs_signatures is installed.

Return type:

bool

acapy_agent.utils.endorsement_setup module

acapy_agent.utils.env module

Environment utility methods.

acapy_agent.utils.env.storage_path(*subpaths, create: bool = False) → Path

Get the default aca-py home directory.

acapy_agent.utils.extract_validation_error module

Extract validation error messages from nested exceptions.

acapy_agent.utils.extract_validation_error.extract_validation_error_message(exc: aiohttp.web.HTTPUnprocessableEntity) → str

Extract marshmallow error message from a nested UnprocessableEntity exception.

acapy_agent.utils.general module

Utility functions for the admin server.

acapy_agent.utils.general.const_compare(string1, string2)

Compare two strings in constant time.

acapy_agent.utils.general.strip_did_prefix(did: str) → str | None

Strip the DID prefix from a DID.

acapy_agent.utils.http module

HTTP utility methods.

exception acapy_agent.utils.http.FetchError(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Error raised when an HTTP fetch fails.

exception acapy_agent.utils.http.PutError(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Error raised when an HTTP put fails.

async acapy_agent.utils.http.fetch(url: str, *, headers: dict | None = None, retry: bool = True, max_attempts: int = 5, interval: float = 1.0, backoff: float = 0.25, request_timeout: float = 10.0, connector: aiohttp.BaseConnector | None = None, session: aiohttp.ClientSession | None = None, json: bool = False)

Fetch from an HTTP server with automatic retries and timeouts.

Parameters:

• url – the address to fetch

• headers – an optional dict of headers to send

• retry – flag to retry the fetch

• max_attempts – the maximum number of attempts to make

• interval – the interval between retries, in seconds

• backoff – the backoff interval, in seconds

• request_timeout – the HTTP request timeout, in seconds

• connector – an optional existing BaseConnector

• session – a shared ClientSession

• json – flag to parse the result as JSON

async acapy_agent.utils.http.fetch_stream(url: str, *, headers: dict | None = None, retry: bool = True, max_attempts: int = 5, interval: float = 1.0, backoff: float = 0.25, request_timeout: float = 10.0, connector: aiohttp.BaseConnector | None = None, session: aiohttp.ClientSession | None = None)

Fetch from an HTTP server with automatic retries and timeouts.

Parameters:

• url – the address to fetch

• headers – an optional dict of headers to send

• retry – flag to retry the fetch

• max_attempts – the maximum number of attempts to make

• interval – the interval between retries, in seconds

• backoff – the backoff interval, in seconds

• request_timeout – the HTTP request timeout, in seconds

• connector – an optional existing BaseConnector

• session – a shared ClientSession

• json – flag to parse the result as JSON

async acapy_agent.utils.http.put_file(url: str, file_data: dict, extra_data: dict, *, retry: bool = True, max_attempts: int = 5, interval: float = 1.0, backoff: float = 0.25, request_timeout: float = 10.0, connector: aiohttp.BaseConnector | None = None, session: aiohttp.ClientSession | None = None, json: bool = False)

Put to HTTP server with automatic retries and timeouts.

Parameters:

• url – the address to use

• file_data – dict with data key and path of file to upload

• extra_data – further content to include in data to put

• headers – an optional dict of headers to send

• retry – flag to retry the fetch

• max_attempts – the maximum number of attempts to make

• interval – the interval between retries, in seconds

• backoff – the backoff interval, in seconds

• request_timeout – the HTTP request timeout, in seconds

• connector – an optional existing BaseConnector

• session – a shared ClientSession

• json – flag to parse the result as JSON

acapy_agent.utils.jwe module

JSON Web Encryption utilities.

class acapy_agent.utils.jwe.B64Value(*args: Any, **kwargs: Any)

Bases: Str

A marshmallow-compatible wrapper for base64-URL values.

class acapy_agent.utils.jwe.JweEnvelope(*, protected: dict | None = None, protected_b64: bytes | None = None, unprotected: dict | None = None, ciphertext: bytes | None = None, iv: bytes | None = None, tag: bytes | None = None, aad: bytes | None = None, with_protected_recipients: bool = False, with_flatten_recipients: bool = True)

Bases: object

JWE envelope instance.

add_recipient(recip: JweRecipient)

Add a recipient to the JWE envelope.

property combined_aad: bytes

Accessor for the additional authenticated data.

classmethod deserialize(message: Mapping[str, Any]) → JweEnvelope

Deserialize a JWE envelope from a mapping.

classmethod from_json(message: bytes | str) → JweEnvelope

Decode a JWE envelope from a JSON string or bytes value.

get_recipient(kid: str) → JweRecipient

Find a recipient by key ID.

property protected_bytes: bytes

Access the protected data encoded as bytes.

This value is used in the additional authenticated data when encrypting.

property recipient_key_ids: Iterable[JweRecipient]

Accessor for an iterator over the JWE recipient key identifiers.

property recipients: Iterable[JweRecipient]

Accessor for an iterator over the JWE recipients.

The headers for each recipient include protected and unprotected headers from the outer envelope.

property recipients_json: List[Dict[str, Any]]

Encode the current recipients for JSON.

serialize() → dict

Serialize the JWE envelope to a mapping.

set_payload(ciphertext: bytes, iv: bytes, tag: bytes, aad: bytes | None = None)

Set the payload of the JWE envelope.

set_protected(protected: Mapping[str, Any])

Set the protected headers of the JWE envelope.

to_json() → str

Serialize the JWE envelope to a JSON string.

class acapy_agent.utils.jwe.JweRecipient(*, encrypted_key: bytes, header: dict | None = None)

Bases: object

A single message recipient.

classmethod deserialize(entry: Mapping[str, Any]) → JweRecipient

Deserialize a JWE recipient from a mapping.

serialize() → dict

Serialize the JWE recipient to a mapping.

class acapy_agent.utils.jwe.JweRecipientSchema(*args: Any, **kwargs: Any)

Bases: Schema

JWE recipient schema.

class acapy_agent.utils.jwe.JweSchema(*args: Any, **kwargs: Any)

Bases: Schema

JWE envelope schema.

acapy_agent.utils.jwe.b64url(value: bytes | str) → str

Encode a string or bytes value as unpadded base64-URL.

acapy_agent.utils.jwe.from_b64url(value: str) → bytes

Decode an unpadded base64-URL value.

acapy_agent.utils.multi_ledger module

Multiledger related utility methods.

acapy_agent.utils.multi_ledger.get_write_ledger_config_for_profile(settings: BaseSettings) → dict

Return initial/default write ledger config on profile creation.

acapy_agent.utils.outofband module

Utilities for creating out-of-band messages.

acapy_agent.utils.outofband.serialize_outofband(message: AgentMessage, did: DIDInfo, endpoint: str) → str

Serialize the agent message as an out-of-band message.

Returns:

An OOB message in URL format.

acapy_agent.utils.profiles module

acapy_agent.utils.repeat module

Utils for repeating tasks.

class acapy_agent.utils.repeat.RepeatAttempt(seq: RepeatSequence, index: int = 1)

Bases: object

Represents the current iteration in a repeat sequence.

property final: bool

Check if this is the last instance in the sequence.

next() → RepeatAttempt

Get the next attempt instance.

property next_interval: float

Calculate the interval before the next attempt.

timeout(interval: float | None = None)

Create a context manager for timing out an attempt.

class acapy_agent.utils.repeat.RepeatSequence(limit: int = 0, interval: float = 0.0, backoff: float = 0.0)

Bases: object

Represents a repetition sequence.

next_interval(index: int) → float

Calculate the time before the next attempt.

start() → RepeatAttempt

Get the first attempt in the sequence.

acapy_agent.utils.server module

Utility functions for server operations in an AcaPy agent.

async acapy_agent.utils.server.remove_unwanted_headers(request, response) → None

Remove unwanted headers from the response.

acapy_agent.utils.stats module

Classes for tracking performance and timing.

class acapy_agent.utils.stats.Collector(*, enabled: bool = True, log_path: str | None = None)

Bases: object

Collector for a set of statistics.

property enabled: bool

Accessor for the collector’s enabled property.

extract(groups: Sequence[str] = None) → dict

Extract statistics for a specific set of groups.

log(name: str, duration: float, start: float | None = None)

Log an entry in the statistics if the collector is enabled.

mark(*names)

Make a custom decorator function for adding to the set of groups.

reset()

Reset the collector’s statistics.

property results: dict

Accessor for the current set of collected statistics.

timer(*groups)

Create a new timer attached to this collector.

wrap(obj, prop_name: str | Sequence[str], groups: Sequence[str] = None, *, ignore_missing: bool = False)

Wrap a method on a class or class instance.

wrap_coro(fn, groups: Sequence[str])

Wrap a coroutine instance to collect timing statistics on execution.

wrap_fn(fn, groups: Sequence[str])

Wrap a function instance to collect timing statistics on execution.

class acapy_agent.utils.stats.Stats

Bases: object

A collection of statistics.

extract(names: Sequence[str] = None) → dict

Summarize the stats in a dictionary.

log(name: str, duration: float)

Log an entry in the stats.

class acapy_agent.utils.stats.Timer(collector: Collector, groups: Sequence[str])

Bases: object

Timer instance for a running task.

classmethod now()

Fetch a standard timer value.

start() → Timer

Start the timer.

stop()

Stop the timer.

acapy_agent.utils.task_queue module

Classes for managing a set of asyncio tasks.

class acapy_agent.utils.task_queue.CompletedTask(task: Task, exc_info: Tuple, ident: str | None = None, timing: dict | None = None)

Bases: object

Represent the result of a queued task.

class acapy_agent.utils.task_queue.PendingTask(coro: Coroutine, complete_hook: Callable | None = None, ident: str | None = None, task_future: Future = None, queued_time: float | None = None)

Bases: object

Represent a task in the queue.

cancel()

Cancel the pending task.

property cancelled

Accessor for the cancelled property.

property task: Task

Accessor for the task.

class acapy_agent.utils.task_queue.TaskQueue(max_active: int = 0, timed: bool = False, trace_fn: Callable | None = None)

Bases: object

A class for managing a set of asyncio tasks.

add_active(task: Task, task_complete: Callable | None = None, ident: str | None = None, timing: dict | None = None) → Task

Register an active async task with an optional completion callback.

Parameters:

• task – The asyncio task instance

• task_complete – An optional callback to run on completion

• ident – A string identifier for the task

• timing – An optional dictionary of timing information

add_pending(pending: PendingTask)

Add a task to the pending queue.

Parameters:

pending – The PendingTask to add to the task queue

cancel()

Cancel any pending or active tasks in the queue.

cancel_pending()

Cancel any pending tasks in the queue.

property cancelled: bool

Accessor for the cancelled property of the queue.

async complete(timeout: float | None = None, cleanup: bool = True)

Cancel any pending tasks and wait for, or cancel active tasks.

completed_task(task: Task, task_complete: Callable, ident: str, timing: dict | None = None)

Clean up after a task has completed and run callbacks.

property current_active: int

Accessor for the current number of active tasks in the queue.

property current_pending: int

Accessor for the current number of pending tasks in the queue.

property current_size: int

Accessor for the total number of tasks in the queue.

drain() → Task

Start the process to run queued tasks.

async flush()

Wait for any active or pending tasks to be completed.

property max_active: int

Accessor for the maximum number of active tasks in the queue.

put(coro: Coroutine, task_complete: Callable | None = None, ident: str | None = None) → PendingTask

Add a new task to the queue, delaying execution if busy.

Parameters:

• coro – The coroutine to run

• task_complete – A callback to run on completion

• ident – A string identifier for the task

Returns: a future resolving to the asyncio task instance once queued

property ready: bool

Accessor for the ready property of the queue.

run(coro: Coroutine, task_complete: Callable | None = None, ident: str | None = None, timing: dict | None = None) → Task

Start executing a coroutine as an async task, bypassing the pending queue.

Parameters:

• coro – The coroutine to run

• task_complete – An optional callback to run on completion

• ident – A string identifier for the task

• timing – An optional dictionary of timing information

Returns: the new asyncio task instance

async wait_for(timeout: float)

Wait for all queued tasks to complete with a timeout.

acapy_agent.utils.task_queue.coro_ident(coro: Coroutine)

Extract an identifier for a coroutine.

async acapy_agent.utils.task_queue.coro_timed(coro: Coroutine, timing: dict)

Capture timing for a coroutine.

acapy_agent.utils.task_queue.task_exc_info(task: Task)

Extract exception info from an asyncio task.

acapy_agent.utils.testing module

acapy_agent.utils.tracing module

Event tracing.

class acapy_agent.utils.tracing.AdminAPIMessageTracingSchema(*args: Any, **kwargs: Any)

Bases: OpenAPISchema

Request/result schema including agent message tracing.

This is to be used as a superclass for aca-py admin input/output messages that need to support tracing.

acapy_agent.utils.tracing.decode_inbound_message(message)

Return bundled message if appropriate.

acapy_agent.utils.tracing.get_timer() → float

Return a timer.

acapy_agent.utils.tracing.trace_event(context, message, handler: str | None = None, outcome: str | None = None, perf_counter: float | None = None, force_trace: bool = False, raise_errors: bool = False) → float

Log a trace event to a configured target.

Parameters:

• context – The application context, attributes of interest are: context[“trace.enabled”]: True if we are logging events context[“trace.target”]: Trace target (“log”, “message” or an http endpoint) context[“trace.tag”]: Tag to be included in trace output

• message – the current message, can be an AgentMessage, InboundMessage, OutboundMessage or Exchange record

• handler (optional) – The handler name for the trace event. If not provided, it defaults to “aca-py.agent”.

• outcome (optional) – The outcome of the trace event.

• perf_counter (optional) – The performance counter value for the trace event.

• force_trace (optional) – If True, forces the trace event to be logged even if tracing is not enabled.

• raise_errors (optional) – If True, raises an exception if there is an error logging the trace event.

Returns:

The value of the performance counter.

Return type:

float

Raises:

Exception – If there is an error logging the trace event and raise_errors is True.

acapy_agent.utils.tracing.tracing_enabled(context, message) → bool

Determine whether to log trace messages or not.