-
Notifications
You must be signed in to change notification settings - Fork 143
Description
What are you really trying to do?
Leverage SandboxImportNotificationPolicy.WARN_ON_UNINTENTIONAL_PASSTHROUGH to identify imports that aren't intentionally passed through to the sandbox.
Describe the bug
Using a sandboxed runner with the workflows defined in a separate from the worker triggers the unintentional passthrough warning. The workflow must be loaded into the sandbox each time and therefore should be exempt from this warning.
Minimal Reproduction
# workflows.py
from datetime import timedelta
from temporalio import workflow
from warning.activities import echo_activity
@workflow.defn
class ExampleWorkflow:
@workflow.run
async def run(self) -> str:
activity_result = await workflow.execute_activity(
echo_activity,
"example",
task_queue="example-task-queue",
start_to_close_timeout=timedelta(seconds=10),
)
return activity_result# activities.py
from temporalio import activity
@activity.defn
async def echo_activity(input: str) -> str:
return f"hello {input}"# worker.py
import asyncio
from temporalio.client import Client
from temporalio.envconfig import ClientConfig
from temporalio.worker import Worker
from temporalio.worker.workflow_sandbox import (
SandboxedWorkflowRunner,
SandboxRestrictions,
)
from temporalio.workflow import SandboxImportNotificationPolicy
from warning.activities import echo_activity
from warning.workflows import ExampleWorkflow
async def main():
config = ClientConfig.load_client_connect_config()
config.setdefault("target_host", "localhost:7233")
# Start client
client = await Client.connect(**config)
worker = Worker(
client,
task_queue="example-task-queue",
workflows=[ExampleWorkflow],
activities=[echo_activity],
workflow_runner=SandboxedWorkflowRunner(
restrictions=SandboxRestrictions.default.with_import_notification_policy(
SandboxImportNotificationPolicy.WARN_ON_DYNAMIC_IMPORT
| SandboxImportNotificationPolicy.WARN_ON_UNINTENTIONAL_PASSTHROUGH
)
),
)
await worker.run()
if __name__ == "__main__":
asyncio.run(main())Starting the above worker yields this log on startup and workflow execution:
uv run warning/worker.py
/.../temporalio/worker/workflow_sandbox/_importer.py:323: UserWarning: Module warning.workflows was not intentionally passed through to the sandbox.
warnings.warn(
/.../temporalio/worker/workflow_sandbox/_importer.py:323: UserWarning: Module warning.activities was not intentionally passed through to the sandbox.
warnings.warn(
Environment/Versions
- Temporal Version: SDK 1.20