aries_cloudagent.admin package

Subpackages

• aries_cloudagent.admin.tests package
  • Submodules
  • aries_cloudagent.admin.tests.test_admin_server module

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, retries: 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.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

Register default routes.

aries_cloudagent.admin.routes.register_module_routes(app: <sphinx.ext.autodoc.importer._MockObject object at 0x7fd6621f3400>)

Register default routes with the webserver.

Eventually this should use dynamic registration based on the currently-selected message families.

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(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.messaging.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])

Bases: aries_cloudagent.admin.base_server.BaseAdminServer

Admin HTTP server class.

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

Add a webhook target.

complete_webhooks()

Wait for all pending webhooks to be dispatched, used in testing.

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

Get the aiohttp application instance.

modules_handler(request: <sphinx.ext.autodoc.importer._MockObject object at 0x7fd6623449b0>)

Request handler for the loaded modules list.

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

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

Perform webserver startup actions.

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

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

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

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, retries: int = None)

Bases: object

Class for managing webhook target information.

topic_filter

Accessor for the target’s topic filter.