aries_cloudagent.admin package

Submodules

aries_cloudagent.admin.base_server module

Abstract admin server interface.

class aries_cloudagent.admin.base_server.BaseAdminServer

Bases: abc.ABC

Admin HTTP server class.

start() → None

Start the webserver.

Raises:AdminSetupError – If there was an error starting the webserver

stop() → None

Stop the webserver.

aries_cloudagent.admin.error module

Admin error classes.

exception aries_cloudagent.admin.error.AdminError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.BaseError

Base class for Admin-related errors.

exception aries_cloudagent.admin.error.AdminSetupError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.admin.error.AdminError

Admin server setup or configuration error.

aries_cloudagent.admin.request_context module

Admin request context class.

A request context provided by the admin server to admin route handlers.

class aries_cloudagent.admin.request_context.AdminRequestContext(profile: aries_cloudagent.core.profile.Profile, *, context: aries_cloudagent.config.injection_context.InjectionContext = None, settings: Mapping[str, object] = None)

Bases: object

Context established by the Conductor and passed into message handlers.

inject(base_cls: Type[InjectType], settings: Mapping[str, object] = None, *, required: bool = True) → Optional[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

injector

Accessor for the associated Injector instance.

profile

Accessor for the associated Profile instance.

session() → aries_cloudagent.core.profile.ProfileSession

Start a new interactive session with no transaction support requested.

settings

Accessor for the context settings.

classmethod test_context(session_inject: dict = None, profile: aries_cloudagent.core.profile.Profile = None) → aries_cloudagent.admin.request_context.AdminRequestContext

Quickly set up a new admin request context for tests.

transaction() → 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.

update_settings(settings: Mapping[str, object])

Update the current scope with additional settings.

aries_cloudagent.admin.server module

Admin server classes.

class aries_cloudagent.admin.server.AdminConfigSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.openapi.OpenAPISchema

Schema for the config endpoint.

config

Used by autodoc_mock_imports.

class aries_cloudagent.admin.server.AdminModulesSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.openapi.OpenAPISchema

Schema for the modules endpoint.

result

Used by autodoc_mock_imports.

class aries_cloudagent.admin.server.AdminResetSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.openapi.OpenAPISchema

Schema for the reset endpoint.

class aries_cloudagent.admin.server.AdminResponder(profile: aries_cloudagent.core.profile.Profile, send: Coroutine[T_co, T_contra, V_co], **kwargs)

Bases: aries_cloudagent.messaging.responder.BaseResponder

Handle outgoing messages from message handlers.

send_fn

Accessor for async function to send outbound message.

send_outbound(message: aries_cloudagent.transport.outbound.message.OutboundMessage) → aries_cloudagent.transport.outbound.status.OutboundSendStatus

Send outbound message.

Parameters:message – The OutboundMessage to be sent

send_webhook(topic: str, payload: dict)

Dispatch a webhook. DEPRECATED: use the event bus instead.

Parameters:

• topic – the webhook topic identifier
• payload – the webhook payload value

class aries_cloudagent.admin.server.AdminServer(host: str, port: int, context: aries_cloudagent.config.injection_context.InjectionContext, root_profile: aries_cloudagent.core.profile.Profile, outbound_message_router: Coroutine[T_co, T_contra, V_co], webhook_router: Callable, conductor_stop: Coroutine[T_co, T_contra, V_co], task_queue: aries_cloudagent.utils.task_queue.TaskQueue = None, conductor_stats: Coroutine[T_co, T_contra, V_co] = None)

Bases: aries_cloudagent.admin.base_server.BaseAdminServer

Admin HTTP server class.

config_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Request handler for the server configuration.

Parameters:request – aiohttp request object Returns:The web response

liveliness_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Request handler for liveliness check.

Parameters:request – aiohttp request object Returns:The web response, always indicating True

make_application() → <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>

Get the aiohttp application instance.

notify_fatal_error()

Set our readiness flags to force a restart (openshift).

on_startup(app: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Perform webserver startup actions.

plugins_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Request handler for the loaded plugins list.

Parameters:request – aiohttp request object Returns:The module list response

readiness_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Request handler for liveliness check.

Parameters:request – aiohttp request object Returns:The web response, indicating readiness for further calls

redirect_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Perform redirect to documentation.

send_webhook(profile: aries_cloudagent.core.profile.Profile, topic: str, payload: dict = None)

Add a webhook to the queue, to send to all registered targets.

shutdown_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Request handler for server shutdown.

Parameters:request – aiohttp request object Returns:The web response (empty production)

start() → None

Start the webserver.

Raises:AdminSetupError – If there was an error starting the webserver

status_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Request handler for the server status information.

Parameters:request – aiohttp request object Returns:The web response

status_reset_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>)

Request handler for resetting the timing statistics.

Parameters:request – aiohttp request object Returns:The web response

stop() → None

Stop the webserver.

websocket_handler(request)

Send notifications to admin client over websocket.

class aries_cloudagent.admin.server.AdminShutdownSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.openapi.OpenAPISchema

Response schema for admin Module.

class aries_cloudagent.admin.server.AdminStatusLivelinessSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.openapi.OpenAPISchema

Schema for the liveliness endpoint.

alive

Used by autodoc_mock_imports.

class aries_cloudagent.admin.server.AdminStatusReadinessSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.openapi.OpenAPISchema

Schema for the readiness endpoint.

ready

Used by autodoc_mock_imports.

class aries_cloudagent.admin.server.AdminStatusSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.openapi.OpenAPISchema

Schema for the status endpoint.

conductor

Used by autodoc_mock_imports.

label

Used by autodoc_mock_imports.

timing

Used by autodoc_mock_imports.

version

Used by autodoc_mock_imports.

aries_cloudagent.admin.server.const_compare(string1, string2)

Compare two strings in constant time.

aries_cloudagent.admin.server.debug_middleware(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>, handler: Coroutine[T_co, T_contra, V_co])

Show request detail in debug log.

aries_cloudagent.admin.server.ready_middleware(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fb056712cd0>, handler: Coroutine[T_co, T_contra, V_co])

Only continue if application is ready to take work.