Skip to main content

Continue-As-New - Python SDK

View Markdown

This page answers the following questions for Python developers:

What is Continue-As-New?​

Continue-As-New lets a Workflow Execution close successfully and creates a new Workflow Execution. You can think of it as a checkpoint when your Workflow gets too long or approaches certain scaling limits.

The new Workflow Execution is in the same chain; it keeps the same Workflow Id but gets a new Run Id and a fresh Event History. It also receives your Workflow's usual parameters.

How to Continue-As-New using the Python SDK​

First, design your Workflow parameters so that you can pass in the "current state" when you Continue-As-New into the next Workflow run. This state is typically set to None for the original caller of the Workflow.

View the source code in the context of the rest of the application code.
@dataclass
class ClusterManagerInput:
    state: Optional[ClusterManagerState] = None
    test_continue_as_new: bool = False

@workflow.run
async def run(self, input: ClusterManagerInput) -> ClusterManagerResult:

The test hook in the above snippet is covered below.

Inside your Workflow, call the continue_as_new() function with the same type. This stops the Workflow right away and starts a new one.

View the source code in the context of the rest of the application code.
workflow.continue_as_new(
  ClusterManagerInput(
      state=self.state,
      test_continue_as_new=input.test_continue_as_new,
  )
)

Considerations for Workflows with Message Handlers​

If you use Updates or Signals, don't call Continue-as-New from the handlers. Instead, wait for your handlers to finish in your main Workflow before you run continue_as_new. See the all_handlers_finished example for guidance.

When is it right to Continue-as-New using the Python SDK?​

Use Continue-as-New when your Workflow might encounter degraded performance or Event History Limits.

Temporal tracks your Workflow's progress against these limits to let you know when you should Continue-as-New. Call workflow.info().is_continue_as_new_suggested() to check if it's time.

How to test Continue-as-New using the Python SDK​

Testing Workflows that naturally Continue-as-New may be time-consuming and resource-intensive. Instead, add a test hook to check your Workflow's Continue-as-New behavior faster in automated tests.

For example, when test_continue_as_new == True, this sample creates a test-only variable called self.max_history_length and sets it to a small value. A helper method in the Workflow checks it each time it considers using Continue-as-New:

View the source code in the context of the rest of the application code.
def should_continue_as_new(self) -> bool:
    if workflow.info().is_continue_as_new_suggested():
        return True
    # For testing
    if (
        self.max_history_length
        and workflow.info().get_current_history_length() > self.max_history_length
    ):
        return True
    return False