How to use the PyZeta AOP framework¶
Imports¶
[1]:
# some standard library imports
from os import remove
from typing import Any, List, Protocol, runtime_checkable
# the core imports for writing your custom logic and the plugin
from pyzeta.framework.aop.analyzers.profiling_advice import ProfilingAdvice
from pyzeta.framework.aop.analyzers.stats import StatsReader
from pyzeta.framework.aop.point_cut import PointCut
from pyzeta.framework.aop.rule import Rule
from pyzeta.framework.aop.aspect import Aspect
from pyzeta.framework.aop.advice import Advice
from pyzeta.framework.initialization.initialization_handler import (
PyZetaInitializationHandler,
)
from pyzeta.framework.ioc.container_provider import ContainerProvider
PyZetaInitializationHandler.initPyZetaServices()
Prepare the Example Class¶
[2]:
class MyClass:
"Simple example class with two methods, one of which is to be profiled."
def __init__(self, attr1: str, attr2: bool) -> None:
"Initialize the example class with example data."
self.attr1 = attr1
self.attr2 = attr2
def method1(self, arg1: int) -> None:
"Print an instance attribute and a value calculated from an argument."
counter = self._count(arg1)
print(f"original method: {self.attr2}, {counter}!")
def _count(self, limit: int) -> None:
"Count stuff to make profiles look more interesting."
counter = 0
for _ in range(limit):
counter += 1
return counter
def method2(self) -> str:
"Return a constant string after printing an instance attribute."
print(f"original method: {self.attr1}!")
return "Hello World!"
def method3(self) -> str:
"Actually the same as `method2` but duplicated for container example."
print(f"original method: {self.attr1}!")
return "Bye World!"
# create an object for later demonstrations;
# aspects apply globally, even to objects created before advice application!
obj = MyClass("PyZeta.MyClass", False)
# verify the original functionality of MyClass.method2
print(obj.method2())
original method: PyZeta.MyClass!
Hello World!
First Example: Use the Pre-Defined Profiling Advice¶
[3]:
# create the advice and a point cut at which to apply it
fileName = "myclass_method1"
profilingAdvice: Advice[None, Any] = ProfilingAdvice(fileName)
pointCut: PointCut = PointCut(".*1")
# remove any previous statistics files
try:
remove(fileName + ProfilingAdvice.extension)
except FileNotFoundError:
print("no previous stats file to remove!")
# combine advice and point cut into a list of rules
rules: List[Rule[None, Any]] = [Rule(pointCut, profilingAdvice)]
# create the aspect from the list of rules
aspect: Aspect[MyClass, None, Any] = Aspect(rules=rules)
# apply the aspect to the example class - profiling of MyClass.method1 is now enabled!
aspect(MyClass)
[4]:
# run MyClass.method1 to record a profile
obj.method1(arg1=1_000_000)
# verify that MyClass.method2 was not affected
print(obj.method2())
# use the static helper to display the profile
print("-" * 50)
StatsReader.printStats(filename=fileName + ProfilingAdvice.extension)
original method: False, 1000000!
original method: PyZeta.MyClass!
Hello World!
--------------------------------------------------
Wed Jul 26 11:41:34 2023 myclass_method1.cprofile
29 function calls in 0.103 seconds
Ordered by: internal time
ncalls tottime percall cumtime percall filename:lineno(function)
1 0.103 0.103 0.103 0.103 3516148056.py:14(_count)
1 0.000 0.000 0.000 0.000 socket.py:613(send)
2 0.000 0.000 0.000 0.000 iostream.py:535(write)
1 0.000 0.000 0.103 0.103 3516148056.py:9(method1)
1 0.000 0.000 0.000 0.000 {built-in method builtins.print}
1 0.000 0.000 0.000 0.000 iostream.py:203(schedule)
1 0.000 0.000 0.000 0.000 threading.py:1185(is_alive)
2 0.000 0.000 0.000 0.000 iostream.py:465(_schedule_flush)
2 0.000 0.000 0.000 0.000 iostream.py:444(_is_master_process)
1 0.000 0.000 0.000 0.000 profiling_advice.py:44(_stop)
1 0.000 0.000 0.000 0.000 threading.py:1118(_wait_for_tstate_lock)
1 0.000 0.000 0.000 0.000 iostream.py:90(_event_pipe)
2 0.000 0.000 0.000 0.000 {built-in method posix.getpid}
1 0.000 0.000 0.000 0.000 {method 'acquire' of '_thread.lock' objects}
2 0.000 0.000 0.000 0.000 {method 'write' of '_io.StringIO' objects}
2 0.000 0.000 0.000 0.000 {built-in method builtins.isinstance}
2 0.000 0.000 0.000 0.000 {method '__exit__' of '_thread.RLock' objects}
2 0.000 0.000 0.000 0.000 {built-in method builtins.len}
1 0.000 0.000 0.000 0.000 {method 'disable' of '_lsprof.Profiler' objects}
1 0.000 0.000 0.000 0.000 threading.py:568(is_set)
1 0.000 0.000 0.000 0.000 {method 'append' of 'collections.deque' objects}
Second Example: Define Custom Advice¶
[5]:
# define two pieces of advice for subsequent application
advice1: Advice[str, Any] = Advice(
lambda *args, **kwargs: print(f"advice1 pre: {args=}, {kwargs=}"),
lambda returnArg, *args, **kwargs: (
f"advice1 post: {returnArg=}, {args=}, {kwargs=}"
),
)
advice2: Advice[str, Any] = Advice(
lambda *args, **kwargs: print(f"advice2 pre: {args=}, {kwargs=}"),
lambda returnArg, *args, **kwargs: (
f"advice2 post: {returnArg=}, {args=}, {kwargs=}"
),
)
# define a point cut that filters for MyClass.method2
pointCut: PointCut = PointCut(".*2")
# combine the pieces of advice and the point cut into a list of rules
rules: List[Rule[str, Any]] = [
Rule(pointCut, advice1),
Rule(pointCut, advice2),
]
# create the aspect from the list of rules
aspect: Aspect[MyClass, str, Any] = Aspect(rules=rules)
# apply the aspect - MyClass.method2 is now wrapped with additional print statements!
aspect(MyClass)
[6]:
# run MyClass.method2 to observe the logic added by the aspect
print(obj.method2())
advice2 pre: args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}
advice1 pre: args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}
original method: PyZeta.MyClass!
advice2 post: returnArg="advice1 post: returnArg='Hello World!', args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}", args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}
Third Example: Use Advice with Containers¶
[7]:
# use the same general setup as in the second example but for method3
advice3: Advice[str, Any] = Advice(
lambda *args, **kwargs: print(f"advice3 pre: {args=}, {kwargs=}"),
lambda returnArg, *args, **kwargs: (
f"advice3 post: {returnArg=}, {args=}, {kwargs=}"
),
)
advice4: Advice[str, Any] = Advice(
lambda *args, **kwargs: print(f"advice4 pre: {args=}, {kwargs=}"),
lambda returnArg, *args, **kwargs: (
f"advice4 post: {returnArg=}, {args=}, {kwargs=}"
),
)
# define a point cut that filters for MyClass.method3
pointCut: PointCut = PointCut(".*3")
# combine the pieces of advice and the point cut into a list of rules
rules: List[Rule[str, Any]] = [
Rule(pointCut, advice3),
Rule(pointCut, advice4),
]
# create the aspect from the list of rules
aspect: Aspect[MyClass, str, Any] = Aspect(rules=rules)
[8]:
# create and interface and register MyClass as its implementation
@runtime_checkable
class MyInterface(Protocol):
def method3(self) -> str:
...
ContainerProvider.getContainer().registerAsSingleton(MyInterface, obj)
assert ContainerProvider.getContainer().tryResolve(MyInterface) is obj, "ERROR"
[9]:
# run MyClass.method3 to observe the logic before adding the aspect
obj = ContainerProvider.getContainer().tryResolve(MyInterface)
print(obj.method3())
# add the aspect
ContainerProvider.registerAspectGlobally(aspect, MyInterface)
# run MyClass.method3 to observe the logic added by the aspect
# note that the instance must be resolved from the container, else no aspect!
obj = ContainerProvider.getContainer().tryResolve(MyInterface)
print(obj.method3())
original method: PyZeta.MyClass!
Bye World!
advice4 pre: args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}
advice3 pre: args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}
original method: PyZeta.MyClass!
advice4 post: returnArg="advice3 post: returnArg='Bye World!', args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}", args=(<__main__.MyClass object at 0x7f63504053d0>,), kwargs={}