Description
Is your feature request related to a problem? Please describe.
Creating activities and decorating them is easy using activy.def in python
Making sure that they are all included in worker launch is more difficult.
Describe the solution you'd like
It would be helpful if a utility was provided that allowed
- extracting activities from a class
- extracting activities from a class instance
- extracting activities from a module
Below is a sample implementation for extraction from class and class instances, assuming async method implementations that uses ast and inspection.
It looks like one could find which methods are decorate by checking if fn.__temporal_activity_definition but that is a private variable name and is not exposed in the temporalio activity.py module. Code that uses __temporal_activity_definition would be simpler and not invoking ast.parse(inspect.getsource(cls)) is preferrable.
import ast
import inspect
import typing
class _MyNodeVisitor(ast.NodeVisitor):
def __init__(self):
self.fn_name_to_decorators: dict[str, set[str]] = {}
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef):
self.fn_name_to_decorators[node.name] = set()
for decorator in node.decorator_list:
print(decorator)
if isinstance(decorator, ast.Call):
# noinspection PyUnresolvedReferences
name = (
decorator.func.attr
if isinstance(decorator.func, ast.Attribute)
else decorator.func.id
)
else:
# noinspection PyUnresolvedReferences
name = (
decorator.value.id + "." + decorator.attr
if isinstance(decorator, ast.Attribute)
else decorator.id
)
self.fn_name_to_decorators[node.name].add(name)
def get_fn_name_to_decorators(self) -> dict[str, set[str]]:
return self.fn_name_to_decorators
class ActivitiesListProvider:
@classmethod
def __get_activities(
cls,
instance: typing.Union[
type["ActivitiesListProvider"], "ActivitiesListProvider"
],
) -> list[typing.Callable]:
visitor = _MyNodeVisitor()
visitor.visit(ast.parse(inspect.getsource(cls)))
fn_name_to_decorators: dict[str, set[str]] = visitor.get_fn_name_to_decorators()
activities = []
for fn_name, decorators in fn_name_to_decorators.items():
if "activity.defn" in decorators:
method = getattr(instance, fn_name)
activities.append(method)
return activities
def get_activities_from_instance(self) -> list[typing.Callable]:
return self.__get_activities(self)
@classmethod
def get_activities_from_cls(cls) -> list[typing.Callable]:
return cls.__get_activities(cls)
And some Tests:
from workflow_metrics.temporal_tools import activities_class
from temporalio import activity
class SomeActivities(activities_class.ActivitiesListProvider):
@activity.defn
async def instance_method_activity(self):
pass
@activity.defn
async def class_method_activity(self):
pass
@staticmethod
@activity.defn
async def static_method_activity():
pass
def test_get_activities_from_cls():
assert SomeActivities.get_activities_from_cls() == [
SomeActivities.instance_method_activity,
SomeActivities.class_method_activity,
SomeActivities.static_method_activity,
]
class ActivitiesClassThatNeedsInstance(activities_class.ActivitiesListProvider):
@activity.defn
async def instance_method_activity(self):
pass
@activity.defn
async def class_method_activity(self):
pass
@staticmethod
@activity.defn
async def static_method_activity():
pass
def test_get_activities_from_instance():
inst = ActivitiesClassThatNeedsInstance()
assert inst.get_activities_from_instance() == [
inst.instance_method_activity,
inst.class_method_activity,
inst.static_method_activity,
]
Additional context
In code that I am working on activities are mainly defined in one module when they are fns and in class methods.