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.plugin_installer module

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.

async wait_for_completion()

Wait for all active tasks to complete with timeout.

This is safer than flush() for testing as it doesn’t try to manage the drain loop, just waits for existing tasks.

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.

acapy_agent.utils.wait_for_active_registry module