Source code for aries_cloudagent.messaging.credential_definitions.routes

"""Credential definition admin routes."""

from asyncio import shield

from aiohttp import web
from aiohttp_apispec import docs, request_schema, response_schema

from marshmallow import fields, Schema

from ...issuer.base import BaseIssuer
from ...ledger.base import BaseLedger
from ...storage.base import BaseStorage

from ..valid import INDY_CRED_DEF_ID, INDY_SCHEMA_ID, INDY_VERSION

from .util import CRED_DEF_TAGS, CRED_DEF_SENT_RECORD_TYPE


[docs]class CredentialDefinitionSendRequestSchema(Schema): """Request schema for schema send request.""" schema_id = fields.Str(description="Schema identifier", **INDY_SCHEMA_ID) support_revocation = fields.Boolean( required=False, description="Revocation supported flag" ) tag = fields.Str( required=False, description="Credential definition identifier tag", default="default", example="default", )
[docs]class CredentialDefinitionSendResultsSchema(Schema): """Results schema for schema send request.""" credential_definition_id = fields.Str( description="Credential definition identifier", **INDY_CRED_DEF_ID )
[docs]class CredentialDefinitionSchema(Schema): """Credential definition schema.""" ver = fields.Str(description="Node protocol version", **INDY_VERSION) ident = fields.Str( description="Credential definition identifier", data_key="id", **INDY_CRED_DEF_ID ) schemaId = fields.Str( description="Schema identifier within credential definition identifier", example=":".join(INDY_CRED_DEF_ID["example"].split(":")[3:-1]), # long or short ) typ = fields.Constant( constant="CL", description="Signature type: CL for Camenisch-Lysyanskaya", data_key="type", example="CL", ) tag = fields.Str( description="Tag within credential definition identifier", example=INDY_CRED_DEF_ID["example"].split(":")[-1], ) value = fields.Dict( description="Credential definition primary and revocation values" )
[docs]class CredentialDefinitionGetResultsSchema(Schema): """Results schema for schema get request.""" credential_definition = fields.Nested(CredentialDefinitionSchema)
[docs]class CredentialDefinitionsCreatedResultsSchema(Schema): """Results schema for cred-defs-created request.""" credential_definition_ids = fields.List( fields.Str(description="Credential definition identifiers", **INDY_CRED_DEF_ID) )
[docs]@docs( tags=["credential-definition"], summary="Sends a credential definition to the ledger", ) @request_schema(CredentialDefinitionSendRequestSchema()) @response_schema(CredentialDefinitionSendResultsSchema(), 200) async def credential_definitions_send_credential_definition(request: web.BaseRequest): """ Request handler for sending a credential definition to the ledger. Args: request: aiohttp request object Returns: The credential definition identifier """ context = request.app["request_context"] body = await request.json() schema_id = body.get("schema_id") support_revocation = bool(body.get("support_revocation")) tag = body.get("tag") ledger: BaseLedger = await context.inject(BaseLedger) issuer: BaseIssuer = await context.inject(BaseIssuer) async with ledger: credential_definition_id, credential_definition = await shield( ledger.create_and_send_credential_definition( issuer, schema_id, signature_type=None, tag=tag, support_revocation=support_revocation, ) ) return web.json_response({"credential_definition_id": credential_definition_id})
[docs]@docs( tags=["credential-definition"], parameters=[ { "name": tag, "in": "query", "schema": {"type": "string", "pattern": pat}, "required": False, } for (tag, pat) in CRED_DEF_TAGS.items() ], summary="Search for matching credential definitions that agent originated", ) @response_schema(CredentialDefinitionsCreatedResultsSchema(), 200) async def credential_definitions_created(request: web.BaseRequest): """ Request handler for retrieving credential definitions that current agent created. Args: request: aiohttp request object Returns: The identifiers of matching credential definitions. """ context = request.app["request_context"] storage = await context.inject(BaseStorage) found = await storage.search_records( type_filter=CRED_DEF_SENT_RECORD_TYPE, tag_query={ tag: request.query[tag] for tag in CRED_DEF_TAGS if tag in request.query }, ).fetch_all() return web.json_response( {"credential_definition_ids": [record.value for record in found]} )
[docs]@docs( tags=["credential-definition"], summary="Gets a credential definition from the ledger", ) @response_schema(CredentialDefinitionGetResultsSchema(), 200) async def credential_definitions_get_credential_definition(request: web.BaseRequest): """ Request handler for getting a credential definition from the ledger. Args: request: aiohttp request object Returns: The credential definition details. """ context = request.app["request_context"] credential_definition_id = request.match_info["id"] ledger: BaseLedger = await context.inject(BaseLedger) async with ledger: credential_definition = await ledger.get_credential_definition( credential_definition_id ) return web.json_response({"credential_definition": credential_definition})
[docs]async def register(app: web.Application): """Register routes.""" app.add_routes( [ web.post( "/credential-definitions", credential_definitions_send_credential_definition, ) ] ) app.add_routes( [web.get("/credential-definitions/created", credential_definitions_created,)] ) app.add_routes( [ web.get( "/credential-definitions/{id}", credential_definitions_get_credential_definition, ) ] )