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, Optional, Type

from ..core.profile import Profile, ProfileSession
from ..config.injector import Injector, InjectType
from ..config.injection_context import InjectionContext
from ..config.settings import Settings
from ..connections.models.conn_record import ConnRecord
from ..transport.inbound.receipt import MessageReceipt
from ..utils.classloader import DeferLoad

from .agent_message import AgentMessage

IN_MEM = DeferLoad("aries_cloudagent.core.in_memory.InMemoryProfile")


[docs]class RequestContext: """Context established by the Conductor and passed into message handlers.""" def __init__( self, profile: Profile, *, context: InjectionContext = None, settings: Mapping[str, object] = None ): """Initialize an instance of RequestContext.""" self._connection_ready = False self._connection_record = None self._context = (context or profile.context).start_scope("request", settings) self._message = None self._message_receipt = None self._profile = profile @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) -> Optional[ConnRecord]: """Accessor for the related connection record.""" return self._connection_record @connection_record.setter def connection_record(self, record: ConnRecord): """Setter for the related connection record. :param record: ConnRecord: """ 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._context.settings.get("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._context.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._context.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._context.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 @property def injector(self) -> Injector: """Accessor for the associated `Injector` instance.""" return self._context.injector @property def profile(self) -> Profile: """Accessor for the associated `Profile` instance.""" return self._profile @property def settings(self) -> Settings: """Accessor for the context settings.""" return self._context.settings
[docs] def session(self) -> ProfileSession: """Start a new interactive session with no transaction support requested.""" return self.profile.session(self._context)
[docs] def transaction(self) -> ProfileSession: """Start a new interactive session with commit and rollback support. If the current backend does not support transactions, then commit and rollback operations of the session will not have any effect. """ return self.profile.transaction(self._context)
[docs] def inject( self, base_cls: Type[InjectType], settings: Mapping[str, object] = None, ) -> InjectType: """Get the provided instance of a given class identifier. Args: cls: The base class to retrieve an instance of settings: An optional mapping providing configuration to the provider Returns: An instance of the base class, or None """ return self._context.inject(base_cls, settings)
[docs] def inject_or( self, base_cls: Type[InjectType], settings: Mapping[str, object] = None, default: Optional[InjectType] = None, ) -> Optional[InjectType]: """Get the provided instance of a given class identifier or default if not found. Args: base_cls: The base class to retrieve an instance of settings: An optional dict providing configuration to the provider default: default return value if no instance is found Returns: An instance of the base class, or None """ return self._context.inject_or(base_cls, settings, default)
[docs] def update_settings(self, settings: Mapping[str, object]): """Update the scope with additional settings.""" self._context.update_settings(settings)
[docs] @classmethod def test_context(cls) -> "RequestContext": """Quickly set up a new request context for tests.""" return RequestContext(IN_MEM.resolved.test_profile())
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))