Structuring sandbox friendly activity imports when using the Python SDK

Hi Everyone,

I’m new to Temporal and have been using it with the Python SDK. Its been a great experience so far :slight_smile:

I’ve just updated to version 0.1b3 of the Python SDK and the sandbox is complaining. For example:

import sniffio
RuntimeError: Failed valiating workflow <workflow name>

Currently I have the Workflows importing Activities. The Activities then import multiple libraries.

If I put the imports inside each Activity function the workflow validates and runs correctly.

Does anyone have any Python best practices for structuring the imports to minimise what is imported from Activities into Workflows?


Unfortunately in sniffio’s case, this is hitting the known extending-restricted-classes limitation because that library has a class extending threading.local (also came up for a recent sniffio user on #python-sdk Slack).

We plan on fixing this extending-restricted-classes limitation, but in the meantime, you’ll have to mark that import as pass through. And really you should do this to all non-stdlib-third-party imports you know are side effect free on import (you will save memory/perf compared to having the sandbox reload these modules).

So for sniffio, it’d be something like this when creating your worker:

my_restrictions = dataclasses.replace(
    passthrough_modules=SandboxRestrictions.passthrough_modules_default | SandboxMatcher(access={"sniffio"}),
my_worker = Worker(..., workflow_runner=SandboxedWorkflowRunner(restrictions=my_restrictions))

We are going to make this simpler in the near future, probably with something like:

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    import sniffio
    import requests

Thanks for the quick response! In my case I don’t need sniffio for the Workflow only the Activity.

I read the threads on Slack and the missing piece for me was how to separate the Activity’s imports from the Workflow’s imports while still referencing the Activity from the Workflow.

Here is the structure I’ve come up with:

class MyActivityParams:
    hello: str

async def my_activity(params: MyActivityParams):
    raise NotImplementedError('Definition must not be called')

import sniffio

from my_activity_definition import MyActivityParams

async def my_activity(params: MyActivityParams):
    <call sniffio...>


from my_activity_definition import MyActivityParams
from my_activity_implementation import my_activity

class MyWorkflow:
    async def run(self, params):
        my_params = MyActivityParams(hello='world')
        await workflow.execute_activity(my_activity, my_params, ...)

Is this a reasonable approach or is there another that you would recommend?

That will still call the imports transitively. You can either import what is needed locally in the activity, or you can pass through the sniffio library. I’d recommend the latter for this and all side-effect-free third party libraries. See the post above where I show how to pass through the sniffio library.

I didn’t realise my approach would still result in transitive importing. I’ll pass sniffio through.

Thanks again for your help :slight_smile:

Actually, if you change:

from my_activity_implementation import my_activity


from my_activity_definition import my_activity

It wouldn’t transitively import and it would work