acapy_plugin_toolbox/static_connections.py
async def handle(self, context: RequestContext, responder: BaseResponder):
        """Handle static connection get list request."""
        connection_mgr = ConnectionManager(context)
        wallet: BaseWallet = await context.inject(BaseWallet)
        try:
            tag_filter = dict(
                filter(lambda item: item[1] is not None, {
                    'my_did': context.message.my_did,
                    'their_did': context.message.their_did,
                }.items())
            )
            post_filter = dict(
                filter(lambda item: item[1] is not None, {
                    'initiator': context.message.initiator,
                    'invitation_key': context.message.invitation_key,
                    'invitation_mode': ConnectionRecord.INVITATION_MODE_STATIC,
                    'their_role': context.message.their_role
                }.items())
            )
            records = await ConnectionRecord.query(context, tag_filter, post_filter)
        except StorageNotFoundError:
            report = ProblemReport(
                explain_ltxt='Connection not found.',
                who_retries='none'
            )
            report.assign_thread_from(context.message)
            await responder.send_reply(report)
            return

        def flatten_target(connection, target, my_info):
            """Map for flattening results."""
            return {
                'connection_id': connection.connection_id,
                'their_info': {
                    'label': target.label,
                    'did': target.did,
                    'vk': target.recipient_keys[0],
                    'endpoint': target.endpoint
                },
                'my_info': {
                    'did': my_info.did,
                    'vk': my_info.verkey,
                    'endpoint': context.settings.get("default_endpoint")
                }
            }

        targets = []
        my_info = []
        for record in records:
            targets.append(
                await connection_mgr.get_connection_target(record)
            )
            my_info.append(await wallet.get_local_did(record.my_did))

        results = list(map(
            flatten_target,
            records,
            targets,
            my_info
        ))

        static_connections = StaticConnectionList(results=results)
        static_connections.assign_thread_from(context.message)
        await responder.send_reply(static_connections)
Handle static connection get list request.
Handle static connection get list request.
Handle static connection get list request .
async def handle(self, context: RequestContext, responder: BaseResponder):
        connection_mgr = ConnectionManager(context)
        wallet: BaseWallet = await context.inject(BaseWallet)
        try:
            tag_filter = dict(
                filter(lambda item: item[1] is not None, {
                    'my_did': context.message.my_did,
                    'their_did': context.message.their_did,
                }.items())
            )
            post_filter = dict(
                filter(lambda item: item[1] is not None, {
                    'initiator': context.message.initiator,
                    'invitation_key': context.message.invitation_key,
                    'invitation_mode': ConnectionRecord.INVITATION_MODE_STATIC,
                    'their_role': context.message.their_role
                }.items())
            )
            records = await ConnectionRecord.query(context, tag_filter, post_filter)
        except StorageNotFoundError:
            report = ProblemReport(
                explain_ltxt='Connection not found.',
                who_retries='none'
            )
            report.assign_thread_from(context.message)
            await responder.send_reply(report)
            return
        def flatten_target(connection, target, my_info):
            return {
                'connection_id': connection.connection_id,
                'their_info': {
                    'label': target.label,
                    'did': target.did,
                    'vk': target.recipient_keys[0],
                    'endpoint': target.endpoint
                },
                'my_info': {
                    'did': my_info.did,
                    'vk': my_info.verkey,
                    'endpoint': context.settings.get("default_endpoint")
                }
            }
        targets = []
        my_info = []
        for record in records:
            targets.append(
                await connection_mgr.get_connection_target(record)
            )
            my_info.append(await wallet.get_local_did(record.my_did))
        results = list(map(
            flatten_target,
            records,
            targets,
            my_info
        ))
        static_connections = StaticConnectionList(results=results)
        static_connections.assign_thread_from(context.message)
        await responder.send_reply(static_connections)
async def handle ( self , context : RequestContext , responder : BaseResponder ) : connection_mgr = ConnectionManager ( context ) wallet : BaseWallet = await context . inject ( BaseWallet ) try : tag_filter = dict ( filter ( lambda item : item [ 1 ] is not None , { 'my_did' : context . message . my_did , 'their_did' : context . message . their_did , } . items ( ) ) ) post_filter = dict ( filter ( lambda item : item [ 1 ] is not None , { 'initiator' : context . message . initiator , 'invitation_key' : context . message . invitation_key , 'invitation_mode' : ConnectionRecord . INVITATION_MODE_STATIC , 'their_role' : context . message . their_role } . items ( ) ) ) records = await ConnectionRecord . query ( context , tag_filter , post_filter ) except StorageNotFoundError : report = ProblemReport ( explain_ltxt = 'Connection not found.' , who_retries = 'none' ) report . assign_thread_from ( context . message ) await responder . send_reply ( report ) return def flatten_target ( connection , target , my_info ) : """Map for flattening results.""" return { 'connection_id' : connection . connection_id , 'their_info' : { 'label' : target . label , 'did' : target . did , 'vk' : target . recipient_keys [ 0 ] , 'endpoint' : target . endpoint } , 'my_info' : { 'did' : my_info . did , 'vk' : my_info . verkey , 'endpoint' : context . settings . get ( "default_endpoint" ) } } targets = [ ] my_info = [ ] for record in records : targets . append ( await connection_mgr . get_connection_target ( record ) ) my_info . append ( await wallet . get_local_did ( record . my_did ) ) results = list ( map ( flatten_target , records , targets , my_info ) ) static_connections = StaticConnectionList ( results = results ) static_connections . assign_thread_from ( context . message ) await responder . send_reply ( static_connections )
Handle static connection get list request.
Handle static connection get list request .
"""Handle static connection get list request.""" """Map for flattening results."""
self
context
RequestContext
responder
BaseResponder