Creating workflow classes dynamically

Community Support
,
1

Hi Folks,

I am running an evaluation of Temporal (with the python API) to see if matches our requirements. I have a bunch of functions that i’d like to convert to workflows. I could manually rewrite them to adapt them to the Temporal API’s requirements, but i think that would require the development of quite a bit of boilerplate. I was thinking about that i could create a simplified function decorator, which would be able to generate a class around function, register that one with Temporal and then we’re set.

def my_workflow(fn, start_to_close_timeout = timedelta(hours=2)):
    # Unfortunately Temporal cannot register local-classess, so we try differently
    fn_name = getattr(fn, '__name__', None)
    if fn_name is None:
        raise ValueError("Cannot fetch function name for workflow")

    class_name = snake_to_camel_case(fn_name)

    GeneratedWorkflowType = type(class_name, (), { "run": fn })    
    workflow.run(GeneratedWorkflowType.run)
    workflow.defn(GeneratedWorkflowType)    

    # code removed for clarity

I have initially stumbled on the constraint that Temporal is not accepting local classes as workflow classes. Then i tried to create a new class via type() and use that for registration (as presented in the code above.

Unfortunately it’s not working:

Traceback (most recent call last):
  File "/home/andrei/src/temporal-eval/./api_concept.py", line 109, in <module>
    @my_workflow
     ^^^^^^^^^^^^^^
  File "/home/andrei/src/temporal-eval/./api_concept.py", line 72, in drone_workflow
    workflow.defn(GeneratedWorkflowType)
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/andrei/.local/lib/python3.11/site-packages/temporalio/workflow.py", line 152, in defn
    return decorator(cls)
           ^^^^^^^^^^^^^^
  File "/home/andrei/.local/lib/python3.11/site-packages/temporalio/workflow.py", line 142, in decorator
    _Definition._apply_to_class(
  File "/home/andrei/.local/lib/python3.11/site-packages/temporalio/workflow.py", line 1698, in _apply_to_class
    raise ValueError(f"Invalid workflow class: {issues[0]}")
ValueError: Invalid workflow class: @workflow.run method run must be defined on FlyMission

This is where i am blocked at the moment. I am wondering if it could be possible somehow to make this one to work or i am just descending into a neverending rabbit hole. Did anyone tried to generate Workflow types runtime? Any success stories or ideas are welcome.

Best regards
Andras

2

Check this out: