[Feature Request] Inject dependencies into activities

[gregbrowndev]

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

Hi,

This may admittedly be more of a question than a feature request as there might already be a simple solution I missed. But nonetheless, this may be a topic worth documenting for users new to async/temporal.

I'm looking for a way to pass request/job-specific dependencies into the activity functions, such as the DB session. One idea described below uses a factory to create the worker, like you'd see in some other frameworks. The problem is creating a closure around the activities makes it impossible to import those activities into the workflows throwing away type-safety.

Describe the solution you'd like

Here's an example of how I initially tried to solve the issue:

# main.py
def create_worker() -> Worker:
    my_app: MyApp = bootstrap()

    @activity.defn
    async def task1(user_id: uuid.UUID) -> str:
        return my_app.task1(
            user_id=user_id
        )

    @activity.defn
    async def task2(user_id: uuid.UUID) -> str:
        return my_app.task2(
            user_id=user_id
        )

    @activity.defn
    async def task3(user_id: uuid.UUID) -> str:
        return my_app.task3(
            user_id=user_id
        )
    
    client = await Client.connect("http://localhost:7233")
    return Worker(
        client,
        task_queue="my-task-queue",
        workflows=[MyWorkflow],
        activities=[
            task1,
            task2,
            task3,
        ],
    )


async def main():
    worker = create_worker()
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

This is similar to how you might write a Flask API's create_app factory. Another framework, FastAPI, supports async and has full DI around its route handlers.

I don't have a great deal of experience building async apps in Python, so it isn't completely clear to me whether the above solution would even be safe. The MyApp object returned by bootstrap handles transaction management and all the business logic (follows a Hexagonal approach), so it definitely shouldn't be used for more than one activity at a time.

I'm going to feel silly now after writing all of this if the simplest solution is just to call bootstrap within the activity like below:

@activity.defn
async def task1(user_id: uuid.UUID) -> str:
    my_app = bootstrap()
    return my_app.task1(
        user_id=user_id
    )

I've always thought this would be bad to repeatedly throw away the internal DB session and various HTTP/SDK clients after each "request". But I guess that only works in sync frameworks because it processes one request at a time.

Additional context

I think I may have just rubber-ducked the problem to myself, but it would be good to see some guidance on the best practice here.

Thanks!

[gregbrowndev]

Actually, some form of DI probably would be quite useful, as I now have a circular dependency:

  graph TD;
      activities.py-->bootstrap.py;
      bootstrap.py-->adapter.py;
      adapter.py-->workflow.py;
      workflow.py-->activities.py;

Loading

Hopefully, it's clear enough knowing I'm building a hexagonal-style app without seeing all the source code, but I'll try to explain:

• activities.py contains the Temporal activity definitions
• bootstrap.py is a composition root, it builds all the dependencies of MyApp, i.e. each outbound adapter and their dependencies and returns an instance of MyApp
• adapter.py is an outbound adapter that exposes workflow functions to MyApp, e.g. APIs to start the workflow, run signals and queries without the app depending directly on the Temporal workflow
• workflow.py contains the Temporal workflow definition

For now, I think the only way to break the circular dependency is to use the string form of execute_activity.

[gregbrowndev]

Also, it is interesting to find and read the discussion about workflow/activity contracts. For me at least it wouldn't be a problem if the contract didn't include the signal/query decorators, since that is how I plan to expose those functions on the adapter interface to my app anyway:

class IPriceChangeWorkflowAdapter(t.Protocol):
    async def run(self, price_change: PriceChange) -> None:
        ...

    async def query_some_state(self, price_change: PriceChange) -> Foo:
        ...
    
    async def send_some_signal(self, price_change: PriceChange) -> None:
        ...

and the concrete impl would look like:

class PriceChangeWorkflowAdapter(IPriceChangeWorkflowAdapter):
    client: Client

    def __init__(self, client: Client):
        self.client = client

    async def run(self, price_change: PriceChange) -> None:
        price_change_id = str(price_change.price_change_id)
        await self.client.start_workflow(
            PriceChangeWorkflow.run,
            price_change_id,
            id=price_change_id,
            task_queue="my-task-queue",
        )
    
    async def query_some_state(self, price_change: PriceChange) -> Foo:
        # get workflow object
        # send query 

    async def send_some_signal(self, price_change: PriceChange) -> None:
        # get workflow object
        # send signal 

[cretz]

it isn't completely clear to me whether the above solution would even be safe

It is safe.

We are actively working on #52 which will solve your problem here. You'll be able to more-easily register methods on a class instance or a class with __call__ as an activity. Then, you can just store your externally needed state on the class (you can do this today, but it's a bit confusing from the workflow caller side, see discussion at #66).

[vaidik]

I think I am in kind of a similar situation but not able to wrap my head around how to make it work on the basis of @cretz 's response.

My situation is this:

I have a FastAPI app. It is structured like so:

• Controllers (as in MVC) - get executed on an incoming HTTP request. Sanitize the incoming payload and perform the business logic.
• Services - actually wrap the business logic as well as the calls to the database (models). So controllers handle the HTTP request and then call services. Services can do one or more of these things:
  • Wrap all the business logic
  • Perform one or more database operations (CRUD)
  • Execute part of the business logic asynchronously using Temporal by executing a workflow

So for example, there is an HTTP API to create a user and then send a message on slack to someone. Ideally, sending a message to someone/some channel on Slack happens asynchronously using a Temporal workflow, which executes a Temporal activity. Now what I want to achieve is that I want to keep the logic of sending message on Slack also on the service class to keep everything related to creating a user colocated. Here is the pseudo code:

user_service.py:

from create_user_workflow import CreateUserWorkflow

class UserService:

   def __init__(self):
      self.db = get_db()
      self.temporal_client = Client(...)

   async def create_user(self, username):
      user = User(username=username)
      self.db.add(user)
      self.db.commit()
      await self.temporal_client.start_workflow(
         CreateUserWorkflow,
         ...
      )

    def notify_on_slack(self, username):
      ....
      # Do something to send message on slack about the new user

create_user_workflow.py:

from user_service import UserService

@activity.defn(name="notify_on_slack")
async def notify_on_slack(username):
  user_service = UserService()
  user_service.notify_on_slack(username)

@workflow.defn(name="CreateUserWorkflow")
class CreateUserWorkflow:

  @workflow.run
  def run(self, params):
    await workflow.start_activity(
      notify_on_slack
    )

But this leads to Cyclic Import in Python.

My current workaround is that I have just moved the logic of sending a message to Slack from the UserService to the workflow module, but I am not too happy about that.

What's the right way to do this?

[cretz]

@vaidik - feel free to join us on Slack or the forums. It can be a bit hard to get good answers on closed issues. In your case, it doesn't make much sense to put an activity in a workflow file. There are general ways to avoid this kind of circular import in Python, it's not really specific to Temporal (imagine you had FastAPI service A depending on B which depended on C which depended on A, how would you solve?)