`seagrass.hooks`

class seagrass.hooks.ContextManagerHook(cm: Union[AbstractContextManager[C], Callable[[], AbstractContextManager[C]]], nest: bool = False)

Bases: ProtoHook[Optional[AbstractContextManager[C]]]

A hook that wraps around a context manager, calling the context manager’s __enter__ and __exit__ methods whenever the hook is activated.

property current_context_manager: Optional[AbstractContextManager[C]]: Return the current context manager instance being used by the hook.

property is_active: bool: Returns True if an event that uses this hook is currently being executed.

__init__(cm: Union[AbstractContextManager[C], Callable[[], AbstractContextManager[C]]], nest: bool = False) → None

Create a new ContextManagerHook.

Parameters

cm (t.Union[t.ContextManager, t.Callable[[], t.ContextManager]]) – a context manager (i.e., an object with __enter__ and __exit__ methods) or a function that takes no arguments and creates a new context manager.
nest (bool) –
whether successive invocations of the context manager should be nested. If nest=False and another Seagrass event that uses this hook gets called, the prehook and cleanup stages of the hook will be skipped. In general, you should set nest=True if your context manager should be used along the lines of
with cm: # Execute outer event ... with cm: # Execute inner event ... # Continue executing outer event ...
and nest=False if your context manager should be used as
with cm: # Execute outer event ... # Execute inner event ... # Continue executing outer event ...

prehook(event: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → Optional[AbstractContextManager[C]]

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

class seagrass.hooks.CounterHook

Bases: ProtoHook[None]

A Seagrass hook that counts the number of times an event occurs.

Examples:

>>> from seagrass.hooks import CounterHook

>>> hook = CounterHook()

>>> event_a = auditor.create_event("event_a", hooks=[hook])

>>> event_b = auditor.create_event("event_b", hooks=[hook])

>>> with auditor.start_auditing(log_results=True):
...     for _ in range(15):
...         auditor.raise_event("event_a")
...     for _ in range(8):
...         auditor.raise_event("event_b")
{"message": "CounterHook results", "seagrass": {"event": null, "hook": "CounterHook", "hook_ctx": {"event": "event_a", "count": 15}}, "level": "INFO"}
{"message": "CounterHook results", "seagrass": {"event": null, "hook": "CounterHook", "hook_ctx": {"event": "event_b", "count": 8}}, "level": "INFO"}

__init__() → None

prehook(event_name: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → None

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

class seagrass.hooks.FileOpenHook(**kwargs)

Bases: RuntimeAuditHook

An event hook for tracking calls to the Python standard library’s open function.

__init__(**kwargs) → None

Initialize the RuntimeAuditHook.

Parameters

propagate_errors (Optional[bool]) –

if not equal to None, set the propagate_errors attribute of the hook, which controls whether errors raised by sys_hook() are propagated. Hooks for which propagate_errors is False will log errors raised by sys_hook(), but won’t raise them.

By default, hook.propagate_errors = True for RuntimeAuditHooks.

class seagrass.hooks.LoggingHook(prehook_msg: Optional[Callable[[str, Tuple[Any, ...], Dict[str, Any]], Union[str, Tuple[Any, ...]]]] = None, posthook_msg: Optional[Callable[[str, Any], Union[str, Tuple[Any, ...]]]] = None, loglevel: int = 10)

Bases: ProtoHook[None]

A hook that emits a new log whenever it gets called.

__init__(prehook_msg: Optional[Callable[[str, Tuple[Any, ...], Dict[str, Any]], Union[str, Tuple[Any, ...]]]] = None, posthook_msg: Optional[Callable[[str, Any], Union[str, Tuple[Any, ...]]]] = None, loglevel: int = 10) → None

prehook(event_name: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → None

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

posthook(event_name: str, result: Any, context: None) → None

Run the posthook. The posthook is run at the end of the execution of an audited event, after the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
result (Any) – The value that was returned by the event’s wrapped function.
context (C) – The context that was returned by the original call to prehook.

class seagrass.hooks.ProfilerHook(sort_keys: Union[str, int, Tuple[str, ...]] = (), restrictions: Union[str, int, float, Tuple[Union[str, int, float], ...]] = ())

Bases: ProtoHook[bool]

A Seagrass hook that uses the built-in cProfile module to collect and log performance statistics on events.

__init__(sort_keys: Union[str, int, Tuple[str, ...]] = (), restrictions: Union[str, int, float, Tuple[Union[str, int, float], ...]] = ()) → None

Create a new ProfilerHook.

Parameters

sort_keys (Union[K,Tuple[K,...]]) – a key or list of keys to use to sort the output generated by log_results(). The available keys and their meanings are the same as those of pstats.Stats.sort_stats.
restrictions (Union[R,Tuple[R,...]]) – a restriction or list of restrictions to use for the output generated by log_results(). The available restrictions and their meanings are the same as those of pstats.Stats.print_stats.

prehook(event_name: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → bool

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

get_stats(**kwargs) → Optional[Stats]

Return the profiling statistics as a pstats.Stats class.

Parameters: kwargs – Keyword arguments to pass to the constructor of pstats.Stats.
Returns: an instance of pstats.Stats. If no profiling information was collected, this function will return None instead.
Return type: Optional[pstats.Stats]

reset() → None: Reset the internal profiler.

log_results(logger: Logger) → None: Log the results captured by ProfilerHook.

class seagrass.hooks.StackTraceHook(stack_depth: Optional[int] = 10)

Bases: ProtoHook[None]

An audit hook that captures the stack trace of where events are raised and collects statistics about caller locations.

__init__(stack_depth: Optional[int] = 10) → None

prehook(event: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → None

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

class seagrass.hooks.RuntimeAuditHook(sys_hook: Callable[[str, Tuple[Any, ...]], None], propagate_errors: Optional[bool] = None, traceable: bool = False)

Bases: ProtoHook[Optional[str]]

A hook that executes its body as a Python runtime audit hook, in accordance with PEP 578.

Note

RuntimeAuditHook is only supported for Python versions >= 3.8

Examples: in the code below, RuntimeEventCounterHook is a class derived from the RuntimeAuditHook base class that prints every time a runtime audit event is triggered.

>>> import sys

>>> from seagrass.hooks import RuntimeAuditHook

>>> class RuntimeEventCounterHook(RuntimeAuditHook):
...     def __init__(self):
...         super().__init__(self.sys_hook)
...
...     def sys_hook(self, event, args):
...         print(f"Encountered event={event!r} with args={args}")
...

>>> hook = RuntimeEventCounterHook()

>>> @auditor.audit("my_event", hooks=[hook])
... def my_event(*args):
...     sys.audit("sys.my_event", *args)

>>> with auditor.start_auditing():
...     my_event(42, "hello, world")
Encountered event='sys.my_event' with args=(42, 'hello, world')

property is_active: bool: Return whether or not the hook is currently active (i.e., whether a Seagrass event that uses the hook is currently executing.)

property current_event: Optional[str]: Returns the current Seagrass event being executed that’s hooked by this function. If no events using this hook are being executed, current_event is None.

__init__(sys_hook: Callable[[str, Tuple[Any, ...]], None], propagate_errors: Optional[bool] = None, traceable: bool = False) → None

Initialize the RuntimeAuditHook.

Parameters

propagate_errors (Optional[bool]) –

if not equal to None, set the propagate_errors attribute of the hook, which controls whether errors raised by sys_hook() are propagated. Hooks for which propagate_errors is False will log errors raised by sys_hook(), but won’t raise them.

By default, hook.propagate_errors = True for RuntimeAuditHooks.

prehook(event: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → Optional[str]

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

posthook(event: str, result: Any, context: Optional[str]) → None

Run the posthook. The posthook is run at the end of the execution of an audited event, after the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
result (Any) – The value that was returned by the event’s wrapped function.
context (C) – The context that was returned by the original call to prehook.

class seagrass.hooks.TimerHook

Bases: ProtoHook[float]

__init__() → None

prehook(event_name: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → float

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

posthook(event_name: str, result: Any, context: float) → None

Run the posthook. The posthook is run at the end of the execution of an audited event, after the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
result (Any) – The value that was returned by the event’s wrapped function.
context (C) – The context that was returned by the original call to prehook.

class seagrass.hooks.TracingHook(tracefunc: TraceFunc)

Bases: CleanupHook[Tuple[Optional[str], Token, Token]]

Seagrass hook wrapper for tracing functions.

Example: the code snippet below defines a new hook from TracingHook that checks each frame to see if MY_VAR is defined locally, and if it is, it records MY_VAR’s value.

>>> import seagrass

>>> from seagrass.hooks import TracingHook

>>> def tracefunc(frame, event, arg):
...     if "MY_VAR" in frame.f_locals:
...         MY_VAR = frame.f_locals["MY_VAR"]
...         logger = seagrass.get_audit_logger(None)
...         if logger is not None:
...             logger.info(f"Found MY_VAR={MY_VAR!r}")
...     return tracefunc

>>> hook = TracingHook(tracefunc)

>>> @seagrass.audit(seagrass.auto, hooks=[hook])
... def example():
...     MY_VAR = 100
...     MY_VAR = "hello, world!"

>>> with seagrass.start_auditing():
...     example()
{"message": "Found MY_VAR=100", "seagrass": {"event": "example"}, "level": "INFO"}
{"message": "Found MY_VAR='hello, world!'", "seagrass": {"event": "example"}, "level": "INFO"}

class TraceFunc(*args, **kwargs)

Bases: Protocol

A tracing function set by sys.settrace or threading.settrace. The arguments to this function are the same as those to the function accepted by sys.settrace.

__init__(*args, **kwargs)

__init__(tracefunc: TraceFunc) → None

Create a new TracingHook.

Parameters: tracefunc (TraceFunc) – the function that should be used to perform tracing via sys.settrace.

property is_active: bool: Return whether or not the hook is currently active (i.e., whether a Seagrass event that uses the hook is currently executing.)

property current_event: Optional[str]: Return the current Seagrass event that is being executed.

static get_current_tracing_hook(): Get the current global tracing hook.

prehook(event_name: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) → Tuple[Optional[str], Token, Token]

Run the prehook. The prehook is run at the beginning of the execution of an audited event, before the function wrapped by the event is run.

Parameters

event_name (str) – The name of the event that was triggered.
args (Tuple[Any,...]) – A tuple of the arguments passed to the function wrapped by the event.
kwargs (Dict[str,Any]) – A dictionary of the keyword arguments passed to the function wrapped by the event.

Returns

“context” data that can be used by the posthook.

Return type

C

cleanup(event_name: str, context: Tuple[Optional[str], Token, Token], exc: Tuple[Any, ...]) → None

Perform the hook’s cleanup stage. The event_name and context are the same as those used by the posthook function. If an exception was thrown while executing the event it will be provided in the exception argument, otherwise, exception will be None.

If cleanup returns a boolean value, that value is used to decide whether to suppress any exceptions that were raised during event execution.

Return type: Optional[bool]

seagrass.hooks.get_hook_context(default: ~typing.Union[~seagrass.hooks._utils.T, ~seagrass._typing.Missing] = <seagrass._typing.Missing>) → Union[HookContext, T]

Retrieve the logging context provided by an executing hook.

Raises: LookupError – if no hook is currently being executed, and default is not specified.

class seagrass.hooks.HookContext(name: str, args: Dict[str, Any])

Bases: object

Logging context provided by a hook that gets attached to logs.

__init__(name: str, args: Dict[str, Any]) → None: Create a new HookContext instance.

seagrass.hooks

`seagrass.hooks`