Temporal

Temporal is our workflow orchestration engine. We use it for the majority of our backend work.

If you are not familiar with Temporal, here are some quick links to the Temporal documentation. This guide assumes understanding of how Temporal works.

⏳ Temporal Concepts

🐍 Temporal Python SDK

Our Temporal Abstraction

We use a custom Temporal client from modules.temporal.client that provides a simplified interface for executing activities and workflows. This abstraction handles all the low-level Temporal operations for you.

Importing the Temporal client

from slidespeak.temporal import Temporal

Executing Activities

`Temporal.run`

Runs an activity and waits for the result. This is the most common way to call activities from workflows.

from slidespeak.temporal import Temporal
from mymodule.tasks import fetch_url

result = await Temporal.run(fetch_url, url, timeout=30)

Parameters:

activity: The activity function to execute
*args: Arguments to pass to the activity
timeout: Timeout in seconds (default: 60)
retries: Number of retry attempts (default: 3)

Executing Workflows

`Temporal.start_workflow`

Starts a workflow and returns a handle immediately. Does not wait for completion. Used in API endpoints when you want to return immediately.

This should be the default way to start a workflow in most cases.

from slidespeak.temporal import Temporal
from presentation.generation.workflows import GeneratePresentation

workflow_handle = await Temporal.start_workflow(GeneratePresentation.run, params)
task_id = workflow_handle.id  # Return this to frontend for polling

`Temporal.execute_workflow`

Starts a workflow and waits for completion. Returns the result directly.

from slidespeak.temporal import Temporal
from document.ingestion.workflows import ProcessDocument

result = await Temporal.execute_workflow(ProcessDocument.run, document_id)

When to use:

You need the workflow result immediately
You’re calling from an endpoint that can wait

Child Workflows

`Temporal.execute_child_workflow`

Runs a child workflow from within a workflow and waits for completion. Used for workflow composition.

from slidespeak.temporal import Temporal
from document.ingestion.workflows import StoreDocument

result = await Temporal.execute_child_workflow(
    StoreDocument.run,
    document_info,
    task_queue="default",
)

`Temporal.fire_and_forget_child_workflow`

Starts a child workflow without waiting. Used for background tasks like sending emails. These child workflows will continue running even if the parent workflow completes or fails.

from slidespeak.temporal import Temporal
from modules.email.workflows import SendDownloadEmail

await Temporal.fire_and_forget_child_workflow(
    SendDownloadEmail.run,
    user_email,
    presentation_name,
    url,
    response_format,
)

Organization

All workflows must be organized under a workflows/ directory with one workflow per file. Each file should have a descriptive name matching the workflow class.

Required Structure

Directorypresentation/
- Directoryedit/
  - Directorycontent/
    Directoryworkflows/
    __init__.py
    regenerate.py
    translate.py
    slide.py

File Naming

Each workflow file should be named descriptively, matching the workflow class:

✅ regenerate.py → RegenerateContent workflow
✅ write_presentation.py → WritePresentation workflow
✅ send_download_email.py → SendDownloadEmail workflow
❌ pipeline.py → Too generic
❌ workflows.py → Multiple workflows in one file

Example Workflow File

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from slidespeak.temporal import Temporal
    from presentation.edit.content.models import RegenerateContentInput
    from presentation.edit.content.tasks import regenerate_content

@workflow.defn
class RegenerateContent:
    @workflow.run
    async def run(self, regenerate_content_input: RegenerateContentInput) -> dict[str, str]:
        return await Temporal.run(regenerate_content, regenerate_content_input)

Exporting Workflows

Create an __init__.py file that exports all workflows from the directory:

"""All workflows for editing presentation content."""

from presentation.edit.content.workflows.regenerate import RegenerateContent
from presentation.edit.content.workflows.slide import (
    EditPresentationSlide,
    EditPresentationSlideParams,
)
from presentation.edit.content.workflows.translate import TranslatePresentationContent

__all__ = [
    "RegenerateContent",
    "TranslatePresentationContent",
    "EditPresentationSlide",
    "EditPresentationSlideParams",
]

This allows importing workflows from the parent module:

from presentation.edit.content.workflows import RegenerateContent, EditPresentationSlide