nuropb.contexts.context_manager

Module Contents

Classes

NuropbContextManager

This class is a context manager that’s used to manage a transaction’s context relating to an incoming nuropb service message. When a class instance’s method is decorated with the nuropb_context decorator, the context manager is instantiated and injected into the method as a ctx parameter.

Data

logger

_test_token_cache

_test_user_id_cache

API

nuropb.contexts.context_manager.logger

None

nuropb.contexts.context_manager._test_token_cache: Dict[str, Any]

None

nuropb.contexts.context_manager._test_user_id_cache: Dict[str, Any]

None

class nuropb.contexts.context_manager.NuropbContextManager(context: Dict[str, Any], suppress_exceptions: Optional[bool] = True)[source]

This class is a context manager that’s used to manage a transaction’s context relating to an incoming nuropb service message. When a class instance’s method is decorated with the nuropb_context decorator, the context manager is instantiated and injected into the method as a ctx parameter.

The nuropb context manager is both a sync and async context manager. Events can be added to the context manager and will be sent to the service mesh when the context manager successfully exits. If an exception is raised while the context manager is in scope, the exception is recorded and the transaction in context is considered to have failed. Any events added to the context manager are discarded.

Initialization

_suppress_exceptions: bool

None

_nuropb_payload: Dict[str, Any] | None

None

_context: Dict[str, Any]

None

_user_claims: Dict[str, Any] | None

None

_events: List[Dict[str, Any]]

None

_exc_type: Type[BaseException] | None

None

_exec_value: BaseException | None

None

_exc_tb: types.TracebackType | None

None

_started: bool

None

_done: bool

None

property context: Dict[str, Any]
property user_claims: Dict[str, Any] | None
property events: List[Dict[str, Any]]
property error: Dict[str, Any] | None
add_event(event: Dict[str, Any]) → None[source]

Add an event to the context manager. The event will be sent to the service mesh when the context manager exits successfully.

Event format

  • “topic”: “test_topic”,

  • “event”: “test_event_payload”,

  • “context”: {}

Parameters:

event –

Returns:

_handle_context_exit(exc_type: Type[BaseException] | None, exc_value: BaseException | None, exc_tb: types.TracebackType | None) → bool[source]

This method is for customising the behaviour when a context manager exits. It has the same signature as exit or aexit.

If an exception was raised while the context manager was running, the exception information is recorded and content transaction is considered to have failed. If any events were added to the context manager, they are discarded.

__enter__() → Any[source]

This method is called when entering a context manager with a with statement

async __aenter__() → Any[source]
__exit__(exc_type: Type[BaseException] | None, exc_value: BaseException | None, exc_tb: types.TracebackType | None) → bool | None[source]
async __aexit__(exc_type: Type[BaseException] | None, exc_value: BaseException | None, exc_tb: types.TracebackType | None) → bool | None[source]