aries_cloudagent.core package

Subpackages

• aries_cloudagent.core.in_memory package
  • Subpackages
    • aries_cloudagent.core.in_memory.didcomm package
      • Submodules
      • aries_cloudagent.core.in_memory.didcomm.derive_1pu module
      • aries_cloudagent.core.in_memory.didcomm.derive_ecdh module
  • Submodules
  • aries_cloudagent.core.in_memory.profile module

Submodules

aries_cloudagent.core.conductor module

aries_cloudagent.core.dispatcher module

aries_cloudagent.core.error module

Common exception classes.

exception aries_cloudagent.core.error.BaseError(*args, error_code: str = None, **kwargs)

Bases: Exception

Generic exception class which other exceptions should inherit from.

message

Accessor for the error message.

roll_up

Accessor for nested error messages rolled into one line.

For display: aiohttp.web errors truncate after newline.

exception aries_cloudagent.core.error.ProfileDuplicateError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.ProfileError

Profile with the given name already exists.

exception aries_cloudagent.core.error.ProfileError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.BaseError

Base error for profile operations.

exception aries_cloudagent.core.error.ProfileNotFoundError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.ProfileError

Requested profile was not found.

exception aries_cloudagent.core.error.ProfileSessionInactiveError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.ProfileError

Error raised when a profile session is not currently active.

exception aries_cloudagent.core.error.ProtocolDefinitionValidationError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.BaseError

Error raised when there is a problem validating a protocol definition.

exception aries_cloudagent.core.error.ProtocolMinorVersionNotSupported(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.BaseError

Minimum minor version protocol error.

Error raised when protocol support exists but minimum minor version is higher than in @type parameter.

exception aries_cloudagent.core.error.StartupError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.BaseError

Error raised when there is a problem starting the conductor.

aries_cloudagent.core.event_bus module

A simple event bus.

class aries_cloudagent.core.event_bus.Event(topic: str, payload: Any = None)

Bases: object

A simple event object.

payload

Return this event’s payload.

topic

Return this event’s topic.

with_metadata(metadata: aries_cloudagent.core.event_bus.EventMetadata) → aries_cloudagent.core.event_bus.EventWithMetadata

Annotate event with metadata and return EventWithMetadata object.

class aries_cloudagent.core.event_bus.EventBus

Bases: object

A simple event bus implementation.

notify(profile: Profile, event: aries_cloudagent.core.event_bus.Event)

Notify subscribers of event.

Parameters:

• profile (Profile) – context of the event
• event (Event) – event to emit

subscribe(pattern: Pattern, processor: Callable)

Subscribe to an event.

Parameters:

• pattern (Pattern) – compiled regular expression for matching topics
• processor (Callable) – async callable accepting profile and event

unsubscribe(pattern: Pattern, processor: Callable)

Unsubscribe from an event.

This method is idempotent. Repeated calls to unsubscribe will not result in errors.

Parameters:

• pattern (Pattern) – regular expression used to subscribe the processor
• processor (Callable) – processor to unsubscribe

wait_for_event(waiting_profile: Profile, pattern: Pattern, cond: Optional[Callable[[aries_cloudagent.core.event_bus.Event], bool]] = None) → Iterator[Awaitable[aries_cloudagent.core.event_bus.Event]]

Capture an event and retrieve its value.

class aries_cloudagent.core.event_bus.EventMetadata

Bases: tuple

Metadata passed alongside events to add context.

match

Alias for field number 1

pattern

Alias for field number 0

class aries_cloudagent.core.event_bus.EventWithMetadata(topic: str, payload: Any, metadata: aries_cloudagent.core.event_bus.EventMetadata)

Bases: aries_cloudagent.core.event_bus.Event

Event with metadata passed alongside events to add context.

metadata

Return metadata.

class aries_cloudagent.core.event_bus.MockEventBus

Bases: aries_cloudagent.core.event_bus.EventBus

A mock EventBus for testing.

notify(profile: Profile, event: aries_cloudagent.core.event_bus.Event)

Append the event to MockEventBus.events.

aries_cloudagent.core.goal_code_registry module

Handle registration and publication of supported goal codes.

class aries_cloudagent.core.goal_code_registry.GoalCodeRegistry

Bases: object

Goal code registry.

goal_codes_matching_query(query: str) → Sequence[str]

Return a list of goal codes matching a query string.

register_controllers(*controller_sets)

Add new controllers.

Parameters:controller_sets – Mappings of controller to coroutines

aries_cloudagent.core.oob_processor module

Oob message processor and functions.

class aries_cloudagent.core.oob_processor.OobMessageProcessor(inbound_message_router: Callable[[aries_cloudagent.core.profile.Profile, aries_cloudagent.transport.inbound.message.InboundMessage, Optional[bool]], None])

Bases: object

Out of band message processor.

clean_finished_oob_record(profile: aries_cloudagent.core.profile.Profile, message: aries_cloudagent.messaging.agent_message.AgentMessage)

Clean up oob record associated with agent message, if applicable.

find_oob_record_for_inbound_message(context: aries_cloudagent.messaging.request_context.RequestContext) → Optional[aries_cloudagent.protocols.out_of_band.v1_0.models.oob_record.OobRecord]

Find oob record for inbound message.

find_oob_target_for_outbound_message(profile: aries_cloudagent.core.profile.Profile, outbound_message: aries_cloudagent.transport.outbound.message.OutboundMessage) → Optional[aries_cloudagent.connections.models.connection_target.ConnectionTarget]

Find connection target for the outbound message.

get_thread_id(message: Dict[str, Any]) → str

Extract thread id from agent message dict.

handle_message(profile: aries_cloudagent.core.profile.Profile, messages: List[Dict[str, Any]], oob_record: aries_cloudagent.protocols.out_of_band.v1_0.models.oob_record.OobRecord, their_service: Optional[aries_cloudagent.messaging.decorators.service_decorator.ServiceDecorator] = None)

Message handler for inbound messages.

exception aries_cloudagent.core.oob_processor.OobMessageProcessorError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.BaseError

Base error for OobMessageProcessor.

aries_cloudagent.core.plugin_registry module

Handle registration of plugin modules for extending functionality.

class aries_cloudagent.core.plugin_registry.PluginRegistry(blocklist: Iterable[str] = [])

Bases: object

Plugin registry for indexing application plugins.

init_context(context: aries_cloudagent.config.injection_context.InjectionContext)

Call plugin setup methods on the current context.

load_protocol_version(context: aries_cloudagent.config.injection_context.InjectionContext, mod: module, version_definition: dict = None)

Load a particular protocol version.

load_protocols(context: aries_cloudagent.config.injection_context.InjectionContext, plugin: module)

For modules that don’t implement setup, register protocols manually.

plugin_names

Accessor for a list of all plugin modules.

plugins

Accessor for a list of all plugin modules.

post_process_routes(app)

Call route binary file response OpenAPI fixups if applicable.

register_admin_routes(app)

Call route registration methods on the current context.

register_package(package_name: str) → Sequence[module]

Register all modules (sub-packages) under a given package name.

register_plugin(module_name: str) → module

Register a plugin module.

register_protocol_events(context: aries_cloudagent.config.injection_context.InjectionContext)

Call route register_events methods on the current context.

validate_version(version_list, module_name)

Validate version dict format.

aries_cloudagent.core.profile module

Classes for managing profile information within a request context.

class aries_cloudagent.core.profile.Profile(*, context: aries_cloudagent.config.injection_context.InjectionContext = None, name: str = None, created: bool = False)

Bases: abc.ABC

Base abstraction for handling identity-related state.

BACKEND_NAME = None

DEFAULT_NAME = 'default'

backend

Accessor for the backend implementation name.

close()

Close the profile instance.

context

Accessor for the injection context.

created

Accessor for the created flag indicating a new profile.

inject(base_cls: Type[InjectType], settings: Mapping[str, object] = None) → InjectType

Get the provided instance of a given class identifier.

Parameters:

• cls – The base class to retrieve an instance of
• settings – An optional mapping providing configuration to the provider

Returns:

An instance of the base class, or None

inject_or(base_cls: Type[InjectType], settings: Mapping[str, object] = None, default: Optional[InjectType] = None) → Optional[InjectType]

Get the provided instance of a given class identifier or default if not found.

Parameters:

• base_cls – The base class to retrieve an instance of
• settings – An optional dict providing configuration to the provider
• default – default return value if no instance is found

Returns:

An instance of the base class, or None

name

Accessor for the profile name.

notify(topic: str, payload: Any)

Signal an event.

remove()

Remove the profile.

session(context: aries_cloudagent.config.injection_context.InjectionContext = None) → aries_cloudagent.core.profile.ProfileSession

Start a new interactive session with no transaction support requested.

settings

Accessor for scope-specific settings.

transaction(context: aries_cloudagent.config.injection_context.InjectionContext = None) → aries_cloudagent.core.profile.ProfileSession

Start a new interactive session with commit and rollback support.

If the current backend does not support transactions, then commit and rollback operations of the session will not have any effect.

class aries_cloudagent.core.profile.ProfileManager

Bases: abc.ABC

Handle provision and open for profile instances.

open(context: aries_cloudagent.config.injection_context.InjectionContext, config: Mapping[str, Any] = None) → aries_cloudagent.core.profile.Profile

Open an instance of an existing profile.

provision(context: aries_cloudagent.config.injection_context.InjectionContext, config: Mapping[str, Any] = None) → aries_cloudagent.core.profile.Profile

Provision a new instance of a profile.

class aries_cloudagent.core.profile.ProfileManagerProvider

Bases: aries_cloudagent.config.base.BaseProvider

The standard profile manager provider which keys off the selected wallet type.

MANAGER_TYPES = {'askar': 'aries_cloudagent.askar.profile.AskarProfileManager', 'in_memory': 'aries_cloudagent.core.in_memory.InMemoryProfileManager', 'indy': 'aries_cloudagent.indy.sdk.profile.IndySdkProfileManager'}

provide(settings: aries_cloudagent.config.base.BaseSettings, injector: aries_cloudagent.config.base.BaseInjector)

Create the profile manager instance.

class aries_cloudagent.core.profile.ProfileSession(profile: aries_cloudagent.core.profile.Profile, *, context: aries_cloudagent.config.injection_context.InjectionContext = None, settings: Mapping[str, Any] = None)

Bases: abc.ABC

An active connection to the profile management backend.

active

Accessor for the session active state.

commit()

Commit any updates performed within the transaction.

If the current session is not a transaction, then nothing is performed.

context

Accessor for the associated injection context.

inject(base_cls: Type[InjectType], settings: Mapping[str, object] = None) → InjectType

Get the provided instance of a given class identifier.

Parameters:

• cls – The base class to retrieve an instance of
• settings – An optional mapping providing configuration to the provider

Returns:

An instance of the base class, or None

inject_or(base_cls: Type[InjectType], settings: Mapping[str, object] = None, default: Optional[InjectType] = None) → Optional[InjectType]

Get the provided instance of a given class identifier or default if not found.

Parameters:

• base_cls – The base class to retrieve an instance of
• settings – An optional dict providing configuration to the provider
• default – default return value if no instance is found

Returns:

An instance of the base class, or None

is_transaction

Check if the session supports commit and rollback operations.

profile

Accessor for the associated profile instance.

rollback()

Roll back any updates performed within the transaction.

If the current session is not a transaction, then nothing is performed.

settings

Accessor for scope-specific settings.

aries_cloudagent.core.protocol_registry module

Handle registration and publication of supported protocols.

class aries_cloudagent.core.protocol_registry.ProtocolRegistry

Bases: object

Protocol registry for indexing message families.

controllers

Accessor for a list of all protocol controller functions.

create_msg_types_for_minor_version(typesets, version_definition)

Return mapping of message type to module path for minor versions.

Parameters:

• typesets – Mappings of message types to register
• version_definition – Optional version definition dict

Returns:

Typesets mapping

message_types

Accessor for a list of all message types.

parse_type_string(message_type)

Parse message type string and return dict with info.

prepare_disclosed(context: aries_cloudagent.config.injection_context.InjectionContext, protocols: Sequence[str])

Call controllers and return publicly supported message families and roles.

protocols

Accessor for a list of all message protocols.

protocols_matching_query(query: str) → Sequence[str]

Return a list of message protocols matching a query string.

register_controllers(*controller_sets, version_definition=None)

Add new controllers.

Parameters:controller_sets – Mappings of message families to coroutines

register_message_types(*typesets, version_definition=None)

Add new supported message types.

Parameters:

• typesets – Mappings of message types to register
• version_definition – Optional version definition dict

resolve_message_class(message_type: str) → type

Resolve a message_type to a message class.

Given a message type identifier, this method returns the corresponding registered message class.

Parameters:message_type – Message type to resolve Returns:The resolved message class

aries_cloudagent.core.util module

Core utilities and constants.

aries_cloudagent.core.util.get_proto_default_version(def_path: str, major_version: int = 1) → str

Return default protocol version from version_definition.

aries_cloudagent.core.util.get_proto_default_version_from_msg_class(profile: aries_cloudagent.core.profile.Profile, msg_class: type, major_version: int = 1) → str

Return default protocol version from version_definition.

aries_cloudagent.core.util.get_version_def_from_msg_class(profile: aries_cloudagent.core.profile.Profile, msg_class: type, major_version: int = 1)

Return version_definition of a protocol from msg_class.

aries_cloudagent.core.util.get_version_from_message(msg: aries_cloudagent.messaging.agent_message.AgentMessage) → str

Return version from provided AgentMessage.

aries_cloudagent.core.util.get_version_from_message_type(msg_type: str) → str

Return version from provided message_type.

aries_cloudagent.core.util.validate_get_response_version(profile: aries_cloudagent.core.profile.Profile, rec_version: str, msg_class: type) → Tuple[str, Optional[str]]

Return a tuple with version to respond with and warnings.

Process received version and protocol version definition, returns the tuple.

Parameters:

• profile – Profile
• rec_version – received version from message
• msg_class – type

Returns:

Tuple with response version and any warnings