trace - Trace/audit for class instances

Overview

This module provides trace/audit logging for functions or object methods through context-based logging provided by logging module.

The trace logging is performed by traced decorator. You can use this decorator directly, or use TracedMixin class to automatically decorate methods of class instances on creation. Each decorated callable could log messages before execution, after successful execution or on failed execution (when unhandled execption is raised by callable). The trace decorator can automatically add agent and context information, and include parameters passed to callable, execution time, return value, information about raised exception etc. to log messages.

The trace logging is managed by TraceManager, that allows dynamic configuration of traced callables at runtime.

Example

The following program is an example of small but complex enough code that you can use to experiment with code tracing options. The parts relevant to tracing are highlighted in the code by embedded comments.

# test-trace.py
from __future__ import annotations
import logging
from time import monotonic
from decimal import Decimal
from enum import IntEnum, auto
from firebird.base.types import *
from firebird.base.logging import LogLevel, LoggingIdMixin, get_logger
from firebird.base.trace import TracedMixin, add_trace, trace_manager, TraceFlag, traced

class Mood(IntEnum):
    "Agent moods"
    ANGRY = auto()
    SAD = auto()
    NEUTRAL = auto()
    PLEASED = auto()
    HAPPY = auto()

class Person(LoggingIdMixin, TracedMixin): # LOGGING & TRACE
    "Sample virtual human agent"
    def __init__(self, name: str, mood: Mood=Mood.NEUTRAL):
        self.name: str = name
        self.mood: Mood = mood
        self.partners: List[Person] = []
    # >>> LOGGING & TRACE
    @property
    def _logging_id_(self) -> str:
        return f"{self.mood.name} {self.name}"
    # <<< LOGGING & TRACE
    def change_mood(self, offset: int) -> None:
        result = self.mood + offset
        if result < Mood.ANGRY:
            self.mood = Mood.ANGRY
        elif result > Mood.HAPPY:
            self.mood = Mood.HAPPY
        else:
            self.mood = Mood(result)
    def process(self, message: str) -> None:
        msg = message.lower()
        if msg == "what you are doing here":
            self.change_mood(-1)
        if 'awful' in msg:
            self.change_mood(-1)
        if ('nice' in msg) or ('wonderful' in msg) or ('pleased' in msg):
            if self.mood != Mood.ANGRY:
                self.change_mood(1)
        if 'happy' in msg:
            if self.mood != Mood.ANGRY:
                self.change_mood(2)
        if 'very nice' in msg:
            if self.mood != Mood.ANGRY:
                self.change_mood(1)
        if 'get lost' in msg:
            self.change_mood(-2)
        if self.name.lower() in msg:
            if self.mood == Mood.SAD:
                self.change_mood(1)
        if self.name.lower() not in msg:
            if self.mood == Mood.NEUTRAL:
                self.change_mood(-1)
    def process_response(self, to: str, mood: Mood) -> None:
        if to == 'greeting':
            if self.mood == Mood.NEUTRAL:
                if mood > Mood.NEUTRAL:
                    self.mood = Mood.PLEASED
                elif mood == Mood.ANGRY:
                    self.mood = Mood.SAD
            elif self.mood == Mood.SAD:
                if mood == Mood.SAD:
                    self.mood = Mood.NEUTRAL
                elif mood == Mood.HAPPY:
                    self.mood = Mood.ANGRY
            elif self.mood == Mood.ANGRY and mood == Mood.SAD:
                self.mood = Mood.NEUTRAL
        elif to == 'chat':
            if self.mood == Mood.SAD and mood > Mood.NEUTRAL:
                self.mood = Mood.NEUTRAL
            elif self.mood == Mood.ANGRY and mood == Mood.SAD:
                self.mood = Mood.NEUTRAL
            elif self.mood == Mood.PLEASED and mood == Mood.ANGRY:
                self.mood = Mood.NEUTRAL
            elif self.mood == Mood.HAPPY and mood == Mood.ANGRY:
                self.mood = Mood.SAD
        elif to == 'bye':
            if self.mood == Mood.NEUTRAL:
                if mood == Mood.ANGRY:
                    self.mood = Mood.ANGRY
                elif mood > Mood.NEUTRAL:
                    self.mood = Mood.PLEASED
            elif self.mood == Mood.HAPPY and mood == Mood.ANGRY:
                self.mood = Mood.NEUTRAL
    def meet(self, other: Person) -> None:
        self.partners.append(other)
        self.greeting(other)
    def interact(self, other: Person, message: str) -> Mood:
        print(f"[{other.name}] {message}")
        # >>> LOGGING
        get_logger(self, topic='Person').debug(f'Processing "{message}" from [{other.name}]')
        # <<< LOGGING
        self.process(message)
        return self.mood
    def greeting(self, other: Person) -> None:
        if self.mood == Mood.NEUTRAL:
            msg = f"Hi {other.name}, I'm {self.name}"
        elif self.mood == Mood.ANGRY:
            msg = "Hi"
        elif self.mood == Mood.SAD:
            msg = f"Hi {other.name}"
        else:
            msg = f"Hi {other.name}, I'm {self.name}. I'm {self.mood.name} to meet you."
        self.process_response('greeting', other.interact(self, msg))
    def chat(self) -> None:
        for other in self.partners:
            if self.mood == Mood.ANGRY:
                msg = "What you are doing here?"
            elif self.mood == Mood.SAD:
                msg = "The weather is awful today, don't you think?"
            elif self.mood == Mood.NEUTRAL:
                msg = "It's a fine day, don't you think?"
            elif self.mood == Mood.PLEASED:
                msg = "It's a very nice day, don't you think?"
            else:
                msg = "Today is a wonderful day!"
            self.process_response('chat', other.interact(self, msg))
    def bye(self) -> str:
        while self.partners:
            other = self.partners.pop()
            if self.mood == Mood.ANGRY:
                msg = "Get lost!"
            elif self.mood == Mood.SAD:
                msg = "Bye"
            elif self.mood == Mood.NEUTRAL:
                msg = f"Bye, {other.name}."
            elif self.mood == Mood.PLEASED:
                msg = f"See you, {other.name}!"
            else:
                msg = f"Bye, {other.name}. Have a nice day!"
            self.process_response('bye', other.interact(self, msg))
        if self.mood == Mood.ANGRY:
            result = "I hate this meeting!"
        elif self.mood == Mood.SAD:
            result = "It was a waste of time!"
        elif self.mood == Mood.NEUTRAL:
            result = "It was OK."
        elif self.mood == Mood.PLEASED:
            result = "Nice meeting, I think."
        else:
            result = "What a wonderful meeting!"
        return result
    def __repr__(self) -> str:
        return f"{self.name} [{self.mood.name}]"

def meeting(name: str, persons: List[Person]):
    "Simulation of virtual agents meeting"

    for person in persons:
        person.log_context = name

    start = monotonic()
    print("Meeting started...")
    print(f"Attendees: {', '.join(f'{x.name} [{x.mood.name}]' for x in persons)}")

    for person in persons:
        for other in persons:
            if other is not person:
                person.meet(other)

    for person in persons:
        person.chat()

    for person in persons:
        person.bye()

    e = str(Decimal(monotonic() - start))
    print(f"Meeting closed in {e[:e.find('.')+6]} sec.")
    print(f"Outcome: {', '.join(f'{x.name} [{x.mood.name}]' for x in persons)}")


def test_trace(name: str, first: Mood, second: Mood) -> None:
    print(f"- without trace ----------")
    meeting(name, [Person('Alex', first), Person('David', second)])

    print(f"- trace ------------------")
    # >>> TRACE
    add_trace(Person, 'greeting', traced)
    add_trace(Person, 'bye', traced)
    add_trace(Person, 'chat', traced)
    add_trace(Person, 'change_mood', traced)
    add_trace(Person, 'process', traced, with_args=False)
    add_trace(Person, 'process_response', traced)
    # <<< TRACE
    meeting(name, [Person('Alex', first), Person('David', second)])

if __name__ == '__main__':
    # >>> LOGGING
    logger = logging.getLogger()
    logger.setLevel(LogLevel.NOTSET)
    sh = logging.StreamHandler()
    sh.setFormatter(logging.Formatter('%(levelname)-10s: [%(topic)s][%(agent)s][%(context)s] %(message)s'))
    logger.addHandler(sh)
    # <<< LOGGING
    # >>> TRACE
    trace_manager.trace |= TraceFlag.ACTIVE
    trace_manager.trace |= (TraceFlag.FAIL | TraceFlag.BEFORE | TraceFlag.AFTER)
    # <<< TRACE
    test_trace('TEST-1', Mood.SAD, Mood.PLEASED)

Output from sample code:

> python test-trace.py
- without trace ----------
Meeting started...
Attendees: Alex [SAD], David [PLEASED]
[Alex] Hi David
DEBUG     : [Person][PLEASED David][TEST-1] Processing "Hi David" from [Alex]
[David] Hi Alex, I'm David. I'm PLEASED to meet you.
DEBUG     : [Person][SAD Alex][TEST-1] Processing "Hi Alex, I'm David. I'm PLEASED to meet you." from [David]
[Alex] It's a fine day, don't you think?
DEBUG     : [Person][PLEASED David][TEST-1] Processing "It's a fine day, don't you think?" from [Alex]
[David] It's a very nice day, don't you think?
DEBUG     : [Person][NEUTRAL Alex][TEST-1] Processing "It's a very nice day, don't you think?" from [David]
[Alex] Bye, David. Have a nice day!
DEBUG     : [Person][PLEASED David][TEST-1] Processing "Bye, David. Have a nice day!" from [Alex]
[David] Bye, Alex. Have a nice day!
DEBUG     : [Person][HAPPY Alex][TEST-1] Processing "Bye, Alex. Have a nice day!" from [David]
Meeting closed in 0.00071 sec.
Outcome: Alex [HAPPY], David [HAPPY]
- trace ------------------
Meeting started...
Attendees: Alex [SAD], David [PLEASED]
DEBUG     : [trace][SAD Alex][TEST-1] >>> greeting(other=David [PLEASED])
[Alex] Hi David
DEBUG     : [Person][PLEASED David][TEST-1] Processing "Hi David" from [Alex]
DEBUG     : [trace][PLEASED David][TEST-1] >>> process
DEBUG     : [trace][PLEASED David][TEST-1] <<< process[0.00002]
DEBUG     : [trace][SAD Alex][TEST-1] >>> process_response(to='greeting', mood=<Mood.PLEASED: 4>)
DEBUG     : [trace][SAD Alex][TEST-1] <<< process_response[0.00000]
DEBUG     : [trace][SAD Alex][TEST-1] <<< greeting[0.00050]
DEBUG     : [trace][PLEASED David][TEST-1] >>> greeting(other=Alex [SAD])
[David] Hi Alex, I'm David. I'm PLEASED to meet you.
DEBUG     : [Person][SAD Alex][TEST-1] Processing "Hi Alex, I'm David. I'm PLEASED to meet you." from [David]
DEBUG     : [trace][SAD Alex][TEST-1] >>> process
DEBUG     : [trace][SAD Alex][TEST-1] >>> change_mood(offset=1)
DEBUG     : [trace][SAD Alex][TEST-1] <<< change_mood[0.00000]
DEBUG     : [trace][SAD Alex][TEST-1] <<< process[0.00018]
DEBUG     : [trace][PLEASED David][TEST-1] >>> process_response(to='greeting', mood=<Mood.NEUTRAL: 3>)
DEBUG     : [trace][PLEASED David][TEST-1] <<< process_response[0.00000]
DEBUG     : [trace][PLEASED David][TEST-1] <<< greeting[0.00061]
DEBUG     : [trace][NEUTRAL Alex][TEST-1] >>> chat()
[Alex] It's a fine day, don't you think?
DEBUG     : [Person][PLEASED David][TEST-1] Processing "It's a fine day, don't you think?" from [Alex]
DEBUG     : [trace][PLEASED David][TEST-1] >>> process
DEBUG     : [trace][PLEASED David][TEST-1] <<< process[0.00000]
DEBUG     : [trace][NEUTRAL Alex][TEST-1] >>> process_response(to='chat', mood=<Mood.PLEASED: 4>)
DEBUG     : [trace][NEUTRAL Alex][TEST-1] <<< process_response[0.00000]
DEBUG     : [trace][NEUTRAL Alex][TEST-1] <<< chat[0.00045]
DEBUG     : [trace][PLEASED David][TEST-1] >>> chat()
[David] It's a very nice day, don't you think?
DEBUG     : [Person][NEUTRAL Alex][TEST-1] Processing "It's a very nice day, don't you think?" from [David]
DEBUG     : [trace][NEUTRAL Alex][TEST-1] >>> process
DEBUG     : [trace][NEUTRAL Alex][TEST-1] >>> change_mood(offset=1)
DEBUG     : [trace][NEUTRAL Alex][TEST-1] <<< change_mood[0.00000]
DEBUG     : [trace][PLEASED Alex][TEST-1] >>> change_mood(offset=1)
DEBUG     : [trace][PLEASED Alex][TEST-1] <<< change_mood[0.00000]
DEBUG     : [trace][NEUTRAL Alex][TEST-1] <<< process[0.00035]
DEBUG     : [trace][PLEASED David][TEST-1] >>> process_response(to='chat', mood=<Mood.HAPPY: 5>)
DEBUG     : [trace][PLEASED David][TEST-1] <<< process_response[0.00000]
DEBUG     : [trace][PLEASED David][TEST-1] <<< chat[0.00077]
DEBUG     : [trace][HAPPY Alex][TEST-1] >>> bye()
[Alex] Bye, David. Have a nice day!
DEBUG     : [Person][PLEASED David][TEST-1] Processing "Bye, David. Have a nice day!" from [Alex]
DEBUG     : [trace][PLEASED David][TEST-1] >>> process
DEBUG     : [trace][PLEASED David][TEST-1] >>> change_mood(offset=1)
DEBUG     : [trace][PLEASED David][TEST-1] <<< change_mood[0.00000]
DEBUG     : [trace][PLEASED David][TEST-1] <<< process[0.00017]
DEBUG     : [trace][HAPPY Alex][TEST-1] >>> process_response(to='bye', mood=<Mood.HAPPY: 5>)
DEBUG     : [trace][HAPPY Alex][TEST-1] <<< process_response[0.00000]
DEBUG     : [trace][HAPPY Alex][TEST-1] <<< bye[0.00060] Result: What a wonderful meeting!
DEBUG     : [trace][HAPPY David][TEST-1] >>> bye()
[David] Bye, Alex. Have a nice day!
DEBUG     : [Person][HAPPY Alex][TEST-1] Processing "Bye, Alex. Have a nice day!" from [David]
DEBUG     : [trace][HAPPY Alex][TEST-1] >>> process
DEBUG     : [trace][HAPPY Alex][TEST-1] >>> change_mood(offset=1)
DEBUG     : [trace][HAPPY Alex][TEST-1] <<< change_mood[0.00000]
DEBUG     : [trace][HAPPY Alex][TEST-1] <<< process[0.00018]
DEBUG     : [trace][HAPPY David][TEST-1] >>> process_response(to='bye', mood=<Mood.HAPPY: 5>)
DEBUG     : [trace][HAPPY David][TEST-1] <<< process_response[0.00000]
DEBUG     : [trace][HAPPY David][TEST-1] <<< bye[0.00059] Result: What a wonderful meeting!
Meeting closed in 0.00466 sec.
Outcome: Alex [HAPPY], David [HAPPY]

Enums & Flags

TraceFlag

class firebird.base.trace.TraceFlag(value)

Bases: enum.IntFlag

LoggingManager trace/audit flags

ACTIVE = 1
AFTER = 4
BEFORE = 2
DISABLED = 0
FAIL = 8

Functions

add_trace

firebird.base.trace.add_trace(cls, method, decorator=<class 'firebird.base.trace.traced'>, /, *args, **kwargs)

Add/update trace specification for class method.

Return type

None

remove_trace

firebird.base.trace.remove_trace(cls, method)

Remove trace specification for class method.

Return type

None

trace_object

firebird.base.trace.trace_object(obj, *, strict=False)

Instruments object’s methods with decorator according to trace configuration.

Parameters

strict (bool) – Determines the response if the object class is not registered for trace. Raises exception when True, or return the instance as is when False [default].

Return type

Any

Returns

Decorated instance.

Raises

TypeError – When object class is not registered and strict is True.

Trace manager

TraceManager

class firebird.base.trace.TraceManager

Bases: object

Trace manager.

add_trace(cls, method, decorator=<class 'firebird.base.trace.traced'>, /, *args, **kwargs)

Add/update trace specification for class method.

Return type

None

clear()

Removes all trace specifications.

Return type

None

is_registered(cls)

Return True if class is registered

Return type

bool

register(cls)

Register class for trace. Does nothing if class is already registered.

Return type

None

remove_trace(cls, method)

Remove trace specification for class method.

Return type

None

trace_object(obj, *, strict=False)

Instruments object’s methods with decorator according to trace configuration.

Parameters

strict (bool) – Determines the response if the object class is not registered for trace. Raises exception when True, or return the instance as is when False [default].

Return type

Any

Returns

Decorated instance.

Raises

TypeError – When object class is not registered and strict is True.

property trace

Trace flags

Return type

TraceFlag

Trace/audit decorator

traced

class firebird.base.trace.traced(*, agent=Sentinel('DEFAULT'), context=Sentinel('DEFAULT'), topic='trace', msg_before=Sentinel('DEFAULT'), msg_after=Sentinel('DEFAULT'), msg_failed=Sentinel('DEFAULT'), flags=<TraceFlag.DISABLED: 0>, level=<LogLevel.DEBUG: 10>, max_param_length=Sentinel('UNLIMITED'), extra=None, callback=None, has_result=Sentinel('DEFAULT'), with_args=True)

Bases: object

Base decorator for logging of callables, suitable for trace/audit.

It’s not applied on decorated function/method if FBASE_TRACE environment variable is set to False, or if FBASE_TRACE is not defined and __debug__ is False (optimized Python code).

Both positional and keyword arguments of decorated callable are available by name for f-string type message interpolation as dict passed to logger as positional argument.

__init__(*, agent=Sentinel('DEFAULT'), context=Sentinel('DEFAULT'), topic='trace', msg_before=Sentinel('DEFAULT'), msg_after=Sentinel('DEFAULT'), msg_failed=Sentinel('DEFAULT'), flags=<TraceFlag.DISABLED: 0>, level=<LogLevel.DEBUG: 10>, max_param_length=Sentinel('UNLIMITED'), extra=None, callback=None, has_result=Sentinel('DEFAULT'), with_args=True)
Parameters
  • agent (Any) – Agent identification

  • context (Any) – Context identification

  • topic (str) – Trace/audit logging topic

  • msg_before (str) – Trace/audit message logged before decorated function

  • msg_after (str) – Trace/audit message logged after decorated function

  • msg_failed (str) – Trace/audit message logged when decorated function raises an exception

  • flags (TraceFlag) – Trace flags override

  • level (LogLevel) – Logging level for trace/audit messages

  • max_param_length (int) – Max. length of parameters (longer will be trimmed)

  • extra (Optional[Dict]) – Extra data for LogRecord

log_after(logger, params)

Executed after decorated callable

Return type

None

log_before(logger, params)

Executed before decorated callable

Return type

None

log_failed(logger, params)

Executed when decorated callable raises an exception

Return type

None

set_after_msg(fn, sig)

Sets the DEFAULT after message f-string template.

Return type

None

set_before_msg(fn, sig)

Sets the DEFAULT before message f-string template.

Return type

None

set_fail_msg(fn, sig)

Sets the DEFAULT fail message f-string template.

Return type

None

Mixins

TracedMixin

class firebird.base.trace.TracedMixin(*args, **kwargs)

Bases: object

Mixin class that automatically registers descendants for trace and instruments instances on creation.

Globals

firebird.base.trace.trace_manager

Trace manager