[Feature Request] Suggest Providing activities extraction utilities to get them from a class and a module

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.

Start by reading the activity.defn implementation and the activity.py module, especially how activity definitions are stored. Review the requested class, instance, and module extraction cases and the supplied tests; done means a documented utility API with coverage for those cases.