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.

add_webhook_target(target_url: str, topic_filter: Sequence[str] = None, max_attempts: int = None)

Add a webhook target.

remove_webhook_target(target_url: str)

Remove a webhook target.

send_webhook(topic: str, payload: dict)

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

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

aries_cloudagent.admin.server module

Admin server classes.

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

Bases: sphinx.ext.autodoc.importer._MockObject

Schema for the modules endpoint.

result

Used by autodoc_mock_imports.

class aries_cloudagent.admin.server.AdminResponder(context: aries_cloudagent.config.injection_context.InjectionContext, send: Coroutine[T_co, T_contra, V_co], webhook: Coroutine[T_co, T_contra, V_co], **kwargs)

Bases: aries_cloudagent.messaging.responder.BaseResponder

Handle outgoing messages from message handlers.

send_outbound(message: aries_cloudagent.transport.outbound.message.OutboundMessage)

Send outbound message.

Parameters:message – The OutboundMessage to be sent

send_webhook(topic: str, payload: dict)

Dispatch a webhook.

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, outbound_message_router: Coroutine[T_co, T_contra, V_co], webhook_router: Callable, 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.

add_webhook_target(target_url: str, topic_filter: Sequence[str] = None, max_attempts: int = None)

Add a webhook target.

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

Get the aiohttp application instance.

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

Perform webserver startup actions.

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

Request handler for the loaded plugins list.

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

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

Perform redirect to documentation.

remove_webhook_target(target_url: str)

Remove a webhook target.

send_webhook(topic: str, payload: dict)

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

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 0x7ffa2fbf1d30>)

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 0x7ffa2fbf1d30>)

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.AdminStatusSchema(*args, **kwargs)

Bases: sphinx.ext.autodoc.importer._MockObject

Schema for the status endpoint.

class aries_cloudagent.admin.server.WebhookTarget(endpoint: str, topic_filter: Sequence[str] = None, max_attempts: int = None)

Bases: object

Class for managing webhook target information.

topic_filter

Accessor for the target’s topic filter.