Source code for aries_cloudagent.protocols.actionmenu.v1_0.routes
"""Action menu admin routes."""
import logging
from aiohttp import web
from aiohttp_apispec import docs, match_info_schema, request_schema, response_schema
from marshmallow import fields
from ....admin.request_context import AdminRequestContext
from ....connections.models.conn_record import ConnRecord
from ....messaging.models.base import BaseModelError
from ....messaging.models.openapi import OpenAPISchema
from ....messaging.valid import UUIDFour
from ....storage.error import StorageError, StorageNotFoundError
from .messages.menu import Menu, MenuSchema
from .messages.menu_request import MenuRequest
from .messages.perform import Perform
from .models.menu_option import MenuOptionSchema
from .util import MENU_RECORD_TYPE, retrieve_connection_menu, save_connection_menu
LOGGER = logging.getLogger(__name__)
[docs]class PerformRequestSchema(OpenAPISchema):
"""Request schema for performing a menu action."""
name = fields.Str(description="Menu option name", example="Query")
params = fields.Dict(
description=("Input parameter values"),
required=False,
keys=fields.Str(example="parameter"), # marshmallow/apispec v3.0 ignores
values=fields.Str(example=UUIDFour.EXAMPLE),
)
[docs]class MenuJsonSchema(OpenAPISchema):
"""Matches MenuSchema but without the inherited AgentMessage properties."""
title = fields.Str(
required=False,
description="Menu title",
example="My Menu",
)
description = fields.Str(
required=False,
description="Introductory text for the menu",
example="User preferences for window settings",
)
errormsg = fields.Str(
required=False,
description="Optional error message to display in menu header",
example="Error: item not present",
)
options = fields.List(
fields.Nested(MenuOptionSchema),
required=True,
description="List of menu options",
)
[docs]class SendMenuSchema(OpenAPISchema):
"""Request schema for sending a menu to a connection."""
menu = fields.Nested(
MenuJsonSchema(),
required=True,
description="Menu to send to connection",
)
[docs]class MenuConnIdMatchInfoSchema(OpenAPISchema):
"""Path parameters and validators for request taking connection id."""
conn_id = fields.Str(
description="Connection identifier", required=True, example=UUIDFour.EXAMPLE
)
[docs]class ActionMenuFetchResultSchema(OpenAPISchema):
"""Result schema for action-menu fetch."""
result = fields.Nested(MenuSchema, description="Action menu")
[docs]async def register(app: web.Application):
"""Register routes."""
app.add_routes(
[
web.post("/action-menu/{conn_id}/close", actionmenu_close),
web.post("/action-menu/{conn_id}/fetch", actionmenu_fetch),
web.post("/action-menu/{conn_id}/perform", actionmenu_perform),
web.post("/action-menu/{conn_id}/request", actionmenu_request),
web.post("/action-menu/{conn_id}/send-menu", actionmenu_send),
]
)
[docs]def post_process_routes(app: web.Application):
"""Amend swagger API."""
# Add top-level tags description
if "tags" not in app._state["swagger_dict"]:
app._state["swagger_dict"]["tags"] = []
app._state["swagger_dict"]["tags"].append(
{"name": "action-menu", "description": "Menu interaction over connection"}
)