seagrass

class seagrass.Auditor(logger: str = 'seagrass')

Bases: object

An auditing instance that allows you to dynamically audit and profile code.

__init__(logger: str = 'seagrass') → None

Create a new Auditor instance.

Parameters

logger (Union[str,logging.Logger]) – The logger that this auditor should use. When set to a string the auditor uses the logger returned by logging.getLogger(logger).

property enabled: bool

Return whether or not the auditor is enabled.

Type

bool

property event_filter: Callable[[str], bool]

A filter over the events being audited.

Parameters

filter (EventFilter) – the filter that should be added to the auditor. The filter should satisfy the seagrass.events.EventFilter interface – that is, they should be functions that take a single string and return a bool.

Raises

TypeError – if the filters don’t satisfy the EventFilter interface.

Examples:

>>> from seagrass.hooks import LoggingHook

>>> hook = LoggingHook(prehook_msg = lambda event, *args: f"{event} called")

>>> _ = auditor.create_event("example.foo", hooks=[hook])

>>> with auditor.start_auditing():
...     auditor.raise_event("example.foo")
{"message": "example.foo called", "seagrass": {"event": "example.foo"}, "level": "DEBUG"}

>>> auditor.event_filter = lambda event: not event.startswith("example.")

>>> with auditor.start_auditing():
...     auditor.raise_event("example.foo")
reset_filter() → None

Reset the event filter used by the Auditor.

toggle_auditing(mode: bool) → None

Enable or disable auditing.

Parameters

mode (bool) – When set to True, auditing is enabled; when set to False, auditing is disabled.

start_auditing(filter: Optional[Callable[[str], bool]] = None, reset_hooks: bool = False, log_results: bool = False) → Iterator[None]

Create a new context within which the auditor is enabled. You can replicate this functionality by calling toggle_auditing(), e.g.

try:
    auditor.toggle_auditing(True)
    # Put code under audit here
    ...
finally:
    auditor.toggle_auditing(False)

However, using with auditor.start_auditing() in place of auditor.toggle_auditing has some additional benefits too, e.g. it allows you to access the logger for the most recent auditing context using seagrass.get_audit_logger.

Parameters

  • filter (Callable[[str],bool]) – a filter to apply on which events should be audited.

  • log_results (bool) – Log hooks results with log_results() before exiting the auditing context.

  • reset_hooks (bool) – Reset hooks with reset(): before exiting the auditing context.

audit(event_name: Union[str, Callable[[F], str]], func: Optional[F] = None, hooks: Optional[List[ProtoHook]] = None, use_async: bool = False, **kwargs) → Union[AuditDecorator, F]

Wrap a function with a new auditing event. You can call audit either as a function decorator or as a regular method of Auditor.

Parameters

  • event_name (Union[str,Callable[[F],str]]) – the name of the new event, which must be unique. This parameter can either be a string, or it can be a function that takes the audited function and creates a string from it, e.g. event_name = lambda func: f"event.{func.__name__}".

  • use_async (bool) – wrap the function in an asynchronous event. This should be set to True if func is async.

  • func (Optional[Callable]) – the function that should be wrapped in a new event.

  • hooks (Optional[List[ProtoHook]]) – a list of hooks to call whenever the new event is triggered.

  • kwargs – keyword arguments to pass on to Event.__init__.

Examples: create an event over the function json.dumps using wrap:

>>> import json
>>> from seagrass.hooks import CounterHook
>>> hook = CounterHook()
>>> audumps = auditor.audit("audit.json.dumps", json.dumps, hooks=[hook])
>>> setattr(json, "dumps", audumps)
>>> hook.event_counter["audit.json.dumps"]
0
>>> with auditor.start_auditing():
...     json.dumps({"a": 1, "b": 2})
'{"a": 1, "b": 2}'
>>> hook.event_counter["audit.json.dumps"]
1

Here is another example where we call auditor.audit as a decorator for a function add:

from seagrass import Auditor
from seagrass.hooks import CounterHook
auditor = Auditor()

@auditor.audit("event.add", hooks=[CounterHook()])
def add(x, y):
    return x + y
async_audit(event_name: Union[str, Callable[[F], str]], func: Optional[F] = None, hooks: Optional[List[ProtoHook]] = None, **kwargs) → Union[AuditDecorator, F]

Call audit() with use_async=True. See the documentation for audit() for more information.

create_event(event_name: str, **kwargs) → Callable[[...], None]

Create a new “empty” event. When this event is executed, it runs any hooks that are associated with the event, but the function wrapped by the event itself does nothing.

Parameters

  • event_name (str) – the name of the event that should be created. Event names should be unique.

  • kwargs – keyword arguments. The keyword arguments for this function are the same as those for wrap().

Returns

returns a wrapper function around the event that was created.

Example:

>>> from seagrass import Auditor

>>> from seagrass.hooks import CounterHook

>>> auditor = Auditor()

>>> hook = CounterHook()

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

>>> hook.event_counter["my_signal"]
0

>>> with auditor.start_auditing():
...     auditor.raise_event("my_signal")

>>> hook.event_counter["my_signal"]
1
reset_hooks() → None

Reset all of the hooks used by this Auditor.

raise_event(event_name: str, *args, **kwargs) → Any

Trigger an audit event using the input arguments and keyword arguments.

Parameters

  • event_name (str) – the name of the event to be raised.

  • args – arguments to pass to the event.

  • kwargs – keyword arguments to pass to the event.

Returns

returns the output of the event that was called.

Return type

Any

Raises

seagrass.errors.EventNotFoundError – if the auditor can’t find the event with the provided name.

add_hooks(event_name: str, *hooks: ProtoHook) → None

Add new hooks to an auditing event.

Parameters

  • event_name (str) – the name of the event to add the hooks to.

  • hooks (ProtoHook) – the hooks that should be added to the event.

Raises

seagrass.errors.EventNotFoundError – if the auditor can’t find the event with the provided name.

toggle_event(event_name: str, enabled: bool) → None

Enables or disables an auditing event.

Parameters

  • event_name (str) – the name of the event to toggle.

  • enabled (bool) – whether to enable or disabled the event.

Example:

>>> from seagrass.hooks import CounterHook
>>> hook = CounterHook()
>>> @auditor.audit("event.say_hello", hooks=[hook])
... def say_hello(name):
...     return f"Hello, {name}!"
>>> hook.event_counter["event.say_hello"]
0
>>> with auditor.start_auditing():
...     say_hello("Alice")
'Hello, Alice!'
>>> hook.event_counter["event.say_hello"]
1
>>> # Disable the "event.say_hello" event
>>> auditor.toggle_event("event.say_hello", False)
>>> with auditor.start_auditing():
...     # Since event.say_hello is disabled, the following call to
...     # say_hello will not contribute to its event counter.
...     say_hello("Bob")
'Hello, Bob!'
>>> hook.event_counter["event.say_hello"]
1
log_results() → None

Log results stored by hooks by calling log_results on all LogResultsHook hooks.

class seagrass.SeagrassLogFilter(name='')

Bases: Filter

A custom logging.Filter that attaches contextual information about Seagrass events to logs.

filter(record) → bool

Determine if the specified record is to be logged.

Is the specified record to be logged? Returns 0 for no, nonzero for yes. If deemed appropriate, the record may be modified in-place.

seagrass.get_audit_logger(default: ~typing.Union[~seagrass.auditor.T, ~seagrass._typing.Missing] = <seagrass._typing.Missing>) → Union[Logger, T]

Get the logger belonging to the auditor in the current auditing context.

This function only works in auditing contexts created by Auditor.audit(); it will be unable to get the logger for the current auditing context if you use Auditor.toggle_auditing().

Returns

the logger for the most recent auditing context, or default if no auditing context has been created.

Return type

t.Union[logging.Logger,T]

Raises

LookupError – if this function is called outside of an auditing context (and no default is provided).

seagrass.get_current_event(default: ~typing.Union[~seagrass.events.contexts.T, ~seagrass._typing.Missing] = <seagrass._typing.Missing>) → Union[str, T]

Get the current Seagrass event that is being executed.

Raises

LookupError – if no Seagrass event is currently under execution, and default is not specified.

seagrass.global_auditor() → Auditor

Return the global Seagrass auditor.

class seagrass.create_global_auditor(auditor: Optional[Auditor] = None)

Bases: AbstractContextManager[Auditor]

Create a context with a new global Auditor (as returned by the global_auditor() function.) This is useful for when you want to import a module that uses Seagrass but don’t want to add its events to the current global Auditor.

If an Auditor is passed into this function, it will be used as the global auditor within the created context. Otherwise, a new Auditor instance will be created.

Parameters

auditor (Optional[Auditor]) – the seagrass.Auditor instance that should be used as the global auditor. If no auditor is provided, a new one will be created.

>>> import seagrass

>>> from seagrass.hooks import LoggingHook

>>> hook = LoggingHook(prehook_msg=lambda event, *args: f"called {event}")

>>> with seagrass.create_global_auditor() as auditor:
...     @seagrass.audit("my_event", hooks=[hook])
...     def my_event():
...         pass

>>> with seagrass.start_auditing():
...     my_event()

>>> with auditor.start_auditing():
...     my_event()
{"message": "called my_event", "seagrass": {"event": "my_event"}, "level": "DEBUG"}
__init__(auditor: Optional[Auditor] = None) → None
seagrass.auto(func: Callable) → str

Automatically generate an event name for a function being audited.

Examples:

>>> from seagrass import auto

>>> from time import sleep

>>> event = auditor.audit(auto, sleep)

>>> event.__event_name__
'time.sleep'

>>> from pathlib import Path

>>> event = auditor.audit(auto, Path.home)

>>> event.__event_name__
'pathlib.Path.home'
seagrass.audit(event_name: Union[str, Callable[[F], str]], func: Optional[F] = None, hooks: Optional[List[ProtoHook]] = None, use_async: bool = False, **kwargs) → Union[AuditDecorator, F]

Wrap a function with a new auditing event. You can call audit either as a function decorator or as a regular method of Auditor.

Parameters

  • event_name (Union[str,Callable[[F],str]]) – the name of the new event, which must be unique. This parameter can either be a string, or it can be a function that takes the audited function and creates a string from it, e.g. event_name = lambda func: f"event.{func.__name__}".

  • use_async (bool) – wrap the function in an asynchronous event. This should be set to True if func is async.

  • func (Optional[Callable]) – the function that should be wrapped in a new event.

  • hooks (Optional[List[ProtoHook]]) – a list of hooks to call whenever the new event is triggered.

  • kwargs – keyword arguments to pass on to Event.__init__.

Examples: create an event over the function json.dumps using wrap:

>>> import json
>>> from seagrass.hooks import CounterHook
>>> hook = CounterHook()
>>> audumps = auditor.audit("audit.json.dumps", json.dumps, hooks=[hook])
>>> setattr(json, "dumps", audumps)
>>> hook.event_counter["audit.json.dumps"]
0
>>> with auditor.start_auditing():
...     json.dumps({"a": 1, "b": 2})
'{"a": 1, "b": 2}'
>>> hook.event_counter["audit.json.dumps"]
1

Here is another example where we call auditor.audit as a decorator for a function add:

from seagrass import Auditor
from seagrass.hooks import CounterHook
auditor = Auditor()

@auditor.audit("event.add", hooks=[CounterHook()])
def add(x, y):
    return x + y
seagrass.create_event(event_name: str, **kwargs) → Callable[[...], None]

Create a new “empty” event. When this event is executed, it runs any hooks that are associated with the event, but the function wrapped by the event itself does nothing.

Parameters

  • event_name (str) – the name of the event that should be created. Event names should be unique.

  • kwargs – keyword arguments. The keyword arguments for this function are the same as those for wrap().

Returns

returns a wrapper function around the event that was created.

Example:

>>> from seagrass import Auditor

>>> from seagrass.hooks import CounterHook

>>> auditor = Auditor()

>>> hook = CounterHook()

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

>>> hook.event_counter["my_signal"]
0

>>> with auditor.start_auditing():
...     auditor.raise_event("my_signal")

>>> hook.event_counter["my_signal"]
1
seagrass.raise_event(event_name: str, *args, **kwargs) → Any

Trigger an audit event using the input arguments and keyword arguments.

Parameters

  • event_name (str) – the name of the event to be raised.

  • args – arguments to pass to the event.

  • kwargs – keyword arguments to pass to the event.

Returns

returns the output of the event that was called.

Return type

Any

Raises

seagrass.errors.EventNotFoundError – if the auditor can’t find the event with the provided name.

seagrass.toggle_event(event_name: str, enabled: bool) → None

Enables or disables an auditing event.

Parameters

  • event_name (str) – the name of the event to toggle.

  • enabled (bool) – whether to enable or disabled the event.

Example:

>>> from seagrass.hooks import CounterHook
>>> hook = CounterHook()
>>> @auditor.audit("event.say_hello", hooks=[hook])
... def say_hello(name):
...     return f"Hello, {name}!"
>>> hook.event_counter["event.say_hello"]
0
>>> with auditor.start_auditing():
...     say_hello("Alice")
'Hello, Alice!'
>>> hook.event_counter["event.say_hello"]
1
>>> # Disable the "event.say_hello" event
>>> auditor.toggle_event("event.say_hello", False)
>>> with auditor.start_auditing():
...     # Since event.say_hello is disabled, the following call to
...     # say_hello will not contribute to its event counter.
...     say_hello("Bob")
'Hello, Bob!'
>>> hook.event_counter["event.say_hello"]
1
seagrass.toggle_auditing(mode: bool) → None

Enable or disable auditing.

Parameters

mode (bool) – When set to True, auditing is enabled; when set to False, auditing is disabled.

seagrass.start_auditing(self, filter: Optional[Callable[[str], bool]] = None, reset_hooks: bool = False, log_results: bool = False) → Iterator[None]

Create a new context within which the auditor is enabled. You can replicate this functionality by calling toggle_auditing(), e.g.

try:
    auditor.toggle_auditing(True)
    # Put code under audit here
    ...
finally:
    auditor.toggle_auditing(False)

However, using with auditor.start_auditing() in place of auditor.toggle_auditing has some additional benefits too, e.g. it allows you to access the logger for the most recent auditing context using seagrass.get_audit_logger.

Parameters

  • filter (Callable[[str],bool]) – a filter to apply on which events should be audited.

  • log_results (bool) – Log hooks results with log_results() before exiting the auditing context.

  • reset_hooks (bool) – Reset hooks with reset(): before exiting the auditing context.

seagrass.add_hooks(event_name: str, *hooks: ProtoHook) → None

Add new hooks to an auditing event.

Parameters

  • event_name (str) – the name of the event to add the hooks to.

  • hooks (ProtoHook) – the hooks that should be added to the event.

Raises

seagrass.errors.EventNotFoundError – if the auditor can’t find the event with the provided name.

seagrass.reset_hooks() → None

Reset all of the hooks used by this Auditor.