Skip to main content

Continue-As-New - Python SDK

This page answers the following questions for Python developers:

What is Continue-As-New?​

Continue-As-New lets a Workflow Execution close successfully and create 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 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, 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 how with all_handlers_finished.

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

Use Continue-as-New when your Workflow might hit scaling limits:

  • When the Event History grows too large
  • When the Workflow might get too many Updates or Signals

Temporal tracks when you should Continue-as-New. Just 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 takes too much time and resources. 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