[Feature Request] Support executing Workflow directly

[Kilo59]

Is your feature request related to a problem? Please describe.

I'm often frustrated when I have some piece business logic as part of a Temporal Workflow that I also want to execute outside of a Temporal Worker. I would like to be able to avoid duplicating this code as much as possible.

Describe the solution you'd like

Being able to call a Workflow.run method directly rather than only being able to run it via a temporal worker is one way to get around this issue.

However the below example will throw a temporalio.workflow._NotInWorkflowEventLoopError: Not in workflow event loop error.

import asyncio
import dataclasses

from temporalio import activity, workflow


@dataclasses.dataclass
class MyArgs:
    to_greet: str


@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, arg: MyArgs) -> list[str]:
        log: list[str] = []

        hello = await workflow.execute_activity(
            GreetingActivities.say_hello,
            arg.to_greet,
        )
        print(hello)
        log.append(hello)

        goodbye = await workflow.execute_activity(
            GreetingActivities.say_goodbye,
            arg.to_greet,
        )
        print(goodbye)
        log.append(goodbye)
        return log


class GreetingActivities:
    @activity.defn
    async def say_hello(self, to_greet: str) -> str:
        return f"Hello {to_greet}!"

    @activity.defn
    async def say_goodbye(self, to_greet: str) -> str:
        return f"Goodbye {to_greet}!"


if __name__ == "__main__":
    wf = GreetingWorkflow()
    asyncio.run(wf.run(MyArgs(to_greet="world")))

Additional context

I don't particular care about running the workflow per se, I just would like to avoid duplicating the "activity" orchestration logic in the workflow layer.

Also when run this way I don't expect to retain the standard temporal "durable execution" guarantees.

[cretz]

Being able to call a Workflow.run method directly rather than only being able to run it via a temporal worker is one way to get around this issue.

Those calls like workflow.execute_activity(... are durable calls that only make sense when run in a worker-based environment. They make no sense outside of a durable context. It seems the request is to run Temporal without a server. It would not be reasonable to effectively embed all server logic like retries and such into the SDK, nor do we expect the primitives to be "durable optional".

Also when run this way I don't expect to retain the standard temporal "durable execution" guarantees.

The APIs are only designed for use in a durable environment. If you must have a durable and non-durable way to run the same code, you may consider your own abstraction that checks workflow.in_workflow() to fall back to whatever non-durable form you prefer (e.g. your durable-optional form of execute activity could fall back to just invoking with manual retry logic). But Temporal workflow primitives require the server to be present and require durability.

[Kilo59]

@cretz

Yeah (as I said) I don't care about the durable execution when calling the workflow this way.

I assumed I was going to have to create my own abstraction, but I was hoping to avoid that.
For example is there a way to register a standard async event loop as "Workflow EventLoop" so that it doesn't throw this temporalio.workflow._NotInWorkflowEventLoopError: Not in workflow event loop error?

But assuming no, back to creating my own abstraction...
Is there some kind of Workflow activity method registry that I can pull from? That way I could programatically know what activity methods the workflow would call?
I would think something like this might also be useful when testing workflow business logic.

[cretz]

For example is there a way to register a standard async event loop as "Workflow EventLoop" so that it doesn't throw this temporalio.workflow._NotInWorkflowEventLoopError: Not in workflow event loop error?

I am afraid not. The workflow event loop is a specially-crafted event loop to be deterministic and has certain things the rest of the workflow calls expect to be present. You can check workflow.in_workflow() and only call workflow. module calls if that is true.

Is there some kind of Workflow activity method registry that I can pull from? That way I could programatically know what activity methods the workflow would call?

To clarify, are you asking which activities are available to a workflow? Technically you can call any activity. While there are overloads for type-safe activity invocation, technically you can pass any string as the activity name. And you can invoke activities that could be on any other task queue on any other worker (even in a different programming language). So there is no centralized place to determine all activities a workflow could call. You can keep track of the activities you pass to your worker if you want to know which activities are available on that task queue (assuming you're not using dynamic activities).