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.FAIL | TraceFlag.BEFORE | TraceFlag.AFTER)
trace_manager.trace_active = True
# <<< 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]
Trace configuration¶
New in version 1.1.0.
Trace supports configuration based on config
.
Sample configuration file:
[trace]
flags = ACTIVE | FAIL
;flags = ACTIVE | BEFORE | AFTER | FAIL
classes = trace_ChannelManager, trace_Channel, trace_TextIOServiceImpl, trace_PipeServerHandler
[trace_PipeServerHandler]
source = saturnin.core.protocol.fbdp.PipeServerHandler
methods = close, send_ready, send_close
[trace_ChannelManager]
source = saturnin.core.base.ChannelManager
special = trace_defer
[trace_Channel]
source = saturnin.core.base.Channel
methods = send, receive, close, bind, unbind, connect, disconnect
[trace_DealerChannel]
source = saturnin.core.base.DealerChannel
methods = send, receive, close, bind, unbind, connect, disconnect
[trace_SimpleService]
source = saturnin.core.classic.SimpleService
methods = validate, run, initialize, start
with_args = no
[trace_TextIOServiceImpl]
source = saturnin.sdk.micro.textio.service.TextIOServiceImpl
methods = initialize, configure, validate, finalize
with_args = no
[trace_defer]
method = defer
max_param_length = 50
See also
load_config()
, BaseTraceConfig
,
TracedMethodConfig
, TracedClassConfig
and
TraceConfig
.
Enums & Flags¶
Functions¶
- firebird.base.trace.add_trace(cls: Type, method: str, /, *args, **kwargs) None ¶
shortcut for
trace_manager.add_trace()
- firebird.base.trace.remove_trace(cls: Type, method: str) None ¶
shortcut for
trace_manager.remove_trace()
Trace manager¶
- class firebird.base.trace.TraceManager[source]¶
Bases:
object
Trace manager.
- add_trace(cls: Type, method: str, /, *args, **kwargs) None [source]¶
Add/update trace specification for class method.
- clear_flag(flag: TraceFlag) None [source]¶
Clear flag specified by
flag
mask.- Parameters:
flag (TraceFlag) –
- Return type:
None
- load_config(config: ConfigParser, section: str = 'trace') None [source]¶
Update trace from configuration.
- Parameters:
config (ConfigParser) – ConfigParser instance with trace configuration.
section (str) – Name of ConfigParser section that should be used to get trace configuration.
- Return type:
None
Uses
TraceConfig
,TracedClassConfig
andTracedMethodConfig
to process the configuration.Note
Does not
clear()
existing trace specifications.
- register(cls: Type) None [source]¶
Register class for trace.
- Parameters:
cls (Type) – Class to be registered.
- Return type:
None
Does nothing if class is already registered.
- set_flag(flag: TraceFlag) None [source]¶
Set flag specified by
flag
mask.- Parameters:
flag (TraceFlag) –
- Return type:
None
- trace_object(obj: Any, *, strict: bool = False) Any [source]¶
Instruments object’s methods with decorator according to trace configuration.
- Parameters:
- Return type:
Only methods registered with
add_trace()
are instrumented.
Trace/audit decorator¶
- class firebird.base.trace.traced(*, agent: Any = Sentinel('DEFAULT'), context: Any = Sentinel('DEFAULT'), topic: str = 'trace', msg_before: str = Sentinel('DEFAULT'), msg_after: str = Sentinel('DEFAULT'), msg_failed: str = Sentinel('DEFAULT'), flags: TraceFlag = TraceFlag.NONE, level: LogLevel = LogLevel.DEBUG, max_param_length: int = Sentinel('UNLIMITED'), extra: Dict | None = None, callback: Callable[[Any], bool] | None = None, has_result: bool = Sentinel('DEFAULT'), with_args: bool = True)[source]¶
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 ifFBASE_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.- Parameters:
- __init__(*, agent: Any = Sentinel('DEFAULT'), context: Any = Sentinel('DEFAULT'), topic: str = 'trace', msg_before: str = Sentinel('DEFAULT'), msg_after: str = Sentinel('DEFAULT'), msg_failed: str = Sentinel('DEFAULT'), flags: TraceFlag = TraceFlag.NONE, level: LogLevel = LogLevel.DEBUG, max_param_length: int = Sentinel('UNLIMITED'), extra: Dict | None = None, callback: Callable[[Any], bool] | None = None, has_result: bool = Sentinel('DEFAULT'), with_args: bool = True)[source]¶
- 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 (Dict | None) – Extra data for
LogRecord
callback (Callable[[Any], bool] | None) – Callback function that gets the agent identification as argument, and must return True/False indicating whether trace is allowed.
has_result (bool) – Indicator whether function has result value. If True,
_result_
is available for interpolation inmsg_after
. TheDEFAULT
value means, that value for this argument should be decided from function return value annotation.with_args (bool) – If True, function arguments are available for interpolation in
msg_before
.
- log_after(logger: FBLoggerAdapter, params: Dict) None [source]¶
Executed after decorated callable.
- Parameters:
logger (FBLoggerAdapter) –
params (Dict) –
- Return type:
None
- log_before(logger: FBLoggerAdapter, params: Dict) None [source]¶
Executed before decorated callable.
- Parameters:
logger (FBLoggerAdapter) –
params (Dict) –
- Return type:
None
- log_failed(logger: FBLoggerAdapter, params: Dict) None [source]¶
Executed when decorated callable raises an exception.
- Parameters:
logger (FBLoggerAdapter) –
params (Dict) –
- Return type:
None
- set_after_msg(fn: Callable, sig: Signature) None [source]¶
Sets the DEFAULT after message f-string template.
- set_before_msg(fn: Callable, sig: Signature) None [source]¶
Sets the DEFAULT before message f-string template.
- set_fail_msg(fn: Callable, sig: Signature) None [source]¶
Sets the DEFAULT fail message f-string template.
- callback: Callable[[Any], bool]¶
Callback function that gets the agent identification as argument, and must return True/False indicating whether trace is allowed.
- has_result: bool¶
Indicator whether function has result value. If True,
_result_
is available for interpolation inmsg_after
.
- with_args: bool¶
If True, function arguments are available for interpolation in
msg_before
Mixins¶
Globals¶
- firebird.base.trace.trace_manager: TraceManager¶
Trace manager
Trace configuration classes¶
- class firebird.base.trace.BaseTraceConfig(name: str)[source]¶
Bases:
Config
Base configuration for trace.
- Parameters:
name (str) –
- flags: FlagOption¶
Trace flags override
- has_result: BoolOption¶
Indicator whether function has result value
- level: EnumOption¶
Logging level for trace/audit messages
- with_args: BoolOption¶
If True, function arguments are available for interpolation in
msg_before
- class firebird.base.trace.TracedMethodConfig(name: str)[source]¶
Bases:
BaseTraceConfig
Configuration of traced Python method.
- Parameters:
name (str) –
- flags: FlagOption¶
Trace flags override
- has_result: BoolOption¶
Indicator whether function has result value
- level: EnumOption¶
Logging level for trace/audit messages
- with_args: BoolOption¶
If True, function arguments are available for interpolation in
msg_before
- class firebird.base.trace.TracedClassConfig(name: str)[source]¶
Bases:
BaseTraceConfig
Configuration of traced Python class.
- Parameters:
name (str) –
- apply_to_descendants: BoolOption¶
True].
- Type:
Wherher configuration should be applied also to all registered descendant classes [default
- flags: FlagOption¶
Trace flags override
- has_result: BoolOption¶
Indicator whether function has result value
- level: EnumOption¶
Logging level for trace/audit messages
- methods: ListOption¶
Names of traced class methods
- special: ConfigListOption¶
Configuration sections with extended config of traced class methods
- with_args: BoolOption¶
If True, function arguments are available for interpolation in
msg_before
- class firebird.base.trace.TraceConfig(name: str)[source]¶
Bases:
BaseTraceConfig
Trace manager configuration.
- Parameters:
name (str) –
- autoregister: BoolOption¶
True].
- Type:
When True, unregistered classes are registered automatically [default
- classes: ConfigListOption¶
Configuration sections with traced Python classes [required].
- flags: FlagOption¶
Trace flags override
- has_result: BoolOption¶
Indicator whether function has result value
- level: EnumOption¶
Logging level for trace/audit messages
- with_args: BoolOption¶
If True, function arguments are available for interpolation in
msg_before