Hacktoberfest 2026: los issues que los mantenedores marcaron para octubre, abiertos y aptos para principiantes. Explorar issues de Hacktoberfest

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

Abierto
  • ¿Algo no está bien?
#758 6 comentarios 1 reacción 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
5/5
Tiempo estimado
Más de una semana
Aptitud para principiantes
35/100
Tipo de issue
Nueva funcionalidad
Claridad
Bastante claro
Estado de actividad
Estancado
Stack tecnológico
python
Área
developer-experience

Línea de trabajo

Empieza leyendo la implementación de activity.defn y el módulo activity.py, especialmente cómo se almacenan las definiciones de actividad. Revisa los casos solicitados de extracción de clases, instancias y módulos, y las pruebas proporcionadas; el trabajo estará terminado cuando exista una API de utilidades documentada con cobertura para esos casos.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

enhancement
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.

Lenguaje dominante
Python
Estrellas
1.2k
Forks
245
Merge medio
4 d 13 h
PR fusionados (30 d)
50

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de temporalio/sdk-python

Todos los issues de temporalio/sdk-python

Issues similares

Más issues de Python

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.