Source code for aries_cloudagent.messaging.request_context

"""
Request context class.

A request context provides everything required by handlers and other parts
of the system to process a message.
"""

from typing import Mapping

from ..config.injection_context import InjectionContext
from ..connections.models.connection_record import ConnectionRecord
from ..transport.inbound.receipt import MessageReceipt

from .agent_message import AgentMessage


[docs]class RequestContext(InjectionContext): """Context established by the Conductor and passed into message handlers.""" def __init__( self, *, base_context: InjectionContext = None, settings: Mapping[str, object] = None ): """Initialize an instance of RequestContext.""" super().__init__(settings=settings) if base_context: self._injector = base_context._injector self._scope_name = base_context.scope_name self._scopes = base_context._scopes self.start_scope("request") self._connection_ready = False self._connection_record = None self._message = None self._message_receipt = None @property def connection_ready(self) -> bool: """ Accessor for the flag indicating an active connection with the sender. Returns: True if the connection is active, else False """ return self._connection_ready @connection_ready.setter def connection_ready(self, active: bool): """ Setter for the flag indicating an active connection with the sender. Args: active: The new active value """ self._connection_ready = active @property def connection_record(self) -> ConnectionRecord: """Accessor for the related connection record.""" return self._connection_record @connection_record.setter def connection_record(self, record: ConnectionRecord): """Setter for the related connection record. :param record: ConnectionRecord: """ self._connection_record = record @property def default_endpoint(self) -> str: """ Accessor for the default agent endpoint (from agent config). Returns: The default agent endpoint """ return self.settings["default_endpoint"] @default_endpoint.setter def default_endpoint(self, endpoint: str): """ Setter for the default agent endpoint (from agent config). Args: endpoint: The new default endpoint """ self.settings["default_endpoint"] = endpoint @property def default_label(self) -> str: """ Accessor for the default agent label (from agent config). Returns: The default label """ return self.settings["default_label"] @default_label.setter def default_label(self, label: str): """ Setter for the default agent label (from agent config). Args: label: The new default label """ self.settings["default_label"] = label @property def message(self) -> AgentMessage: """ Accessor for the deserialized message instance. Returns: This context's agent message """ return self._message @message.setter def message(self, msg: AgentMessage): """ Setter for the deserialized message instance. Args: msg: This context's new agent message """ self._message = msg @property def message_receipt(self) -> MessageReceipt: """ Accessor for the message receipt information. Returns: This context's message receipt information """ return self._message_receipt @message_receipt.setter def message_receipt(self, receipt: MessageReceipt): """ Setter for the message receipt information. Args: msg: This context's new message receipt information """ self._message_receipt = receipt def __repr__(self) -> str: """ Provide a human readable representation of this object. Returns: A human readable representation of this object """ skip = () items = ( "{}={}".format(k, repr(v)) for k, v in self.__dict__.items() if k not in skip ) return "<{}({})>".format(self.__class__.__name__, ", ".join(items))
[docs] def copy(self) -> "RequestContext": """Produce a copy of the request context instance.""" return super().copy()