← Cloudflare Workflows / workflows / python

Python Workers API

В этом руководстве рассматривается Python Workflows SDK и приведены инструкции по созданию Workflow на Python.

WorkflowEntrypoint это основная точка входа для Workflow на Python. Он расширяет WorkflowEntrypoint класс и реализует run метод.

from workers import WorkflowEntrypoint

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        # steps here

• step.do(name=None, *, concurrent=False, config=None) : декоратор, позволяющий определить шаг в Workflow.

  • name : необязательное имя шага. Если оно не указано, используется имя функции (func.__name__) используется.
  • concurrent : необязательное булево значение, указывающее, могут ли зависимости этого шага выполняться параллельно.
  • config : необязательный WorkflowStepConfig для настройки поведение повторных попыток для конкретного шага. Это передается в виде словаря Python, а затем преобразуется по типу в WorkflowStepConfig объект.

  Все параметры, кроме name являются именованными (keyword-only) аргументами.

Зависимости определяются неявно по имени параметра. Если имя параметра функции шага совпадает с именем ранее объявленной функции шага, её результат передаётся в этот шаг.

Если вы определяете ctx параметр, контекст шага передаётся в этот аргумент.

from workers import WorkflowEntrypoint

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do()
        async def my_first_step():
            # do some work
            return "Hello World!"

        await my_first_step()

Обратите внимание: декоратор не вызывает шаг сам по себе, а лишь возвращает вызываемый объект (callable), с помощью которого можно запустить шаг. Чтобы шаг выполнился, этот объект нужно вызвать.

При возврате состояния из шага убедитесь, что возвращаемое значение можно сериализовать.

• step.sleep(name, duration)
  • name : имя шага.
  • duration : продолжительность приостановки выполнения в виде number в миллисекундах или как WorkflowDuration-совместимая строка.

async def run(self, event, step):
    await step.sleep("my-sleep-step", "10 seconds")

• step.sleep_until(name, timestamp)
  • name : имя шага.
  • timestamp, это datetime.datetime объект или число секунд от начала эпохи Unix, до которого нужно перевести экземпляр workflow в спящий режим.

import datetime

async def run(self, event, step):
    await step.sleep_until("my-sleep-step", datetime.datetime.now() + datetime.timedelta(seconds=10))

• step.wait_for_event(name, event_type, timeout="24 hours")
  • name : имя шага.
  • event_type : тип события, которое нужно ожидать.
  • timeout : тайм-аут для wait_for_event вызов. Тайм-аут по умолчанию составляет 24 часа.

async def run(self, event, step):
    await step.wait_for_event("my-wait-for-event-step", "my-event-type")

event параметр представляет собой словарь, который содержит полезную нагрузку, переданную экземпляру workflow, а также другие метаданные:

• payload : полезная нагрузка, переданная в экземпляр workflow.
• timestamp : отметка времени запуска workflow.
• instanceId : ID текущего экземпляра workflow.
• workflowName : имя workflow.

Семантика Workflows позволяет перехватывать исключения, которые всплывают на верхний уровень.

Перехват конкретных исключений внутри except блок может работать некорректно, поскольку некоторые ошибки Python не воссоздаются как ошибки того же типа при передаче через RPC-уровень.

async def run(self, event, step):
    async def try_step(fn):
        try:
            return await fn()
        except Exception as e:
            print(f"Successfully caught {type(e).__name__}: {e}")

    @step.do("my_failing")
    async def my_failing():
        print("Executing my_failing")
        raise TypeError("Intentional error in my_failing")

    await try_step(my_failing)

Python Workflows SDK предоставляет NonRetryableError класс, который можно использовать, чтобы указать, что шаг не должен повторяться.

from workers.workflows import NonRetryableError

raise NonRetryableError(message)

Вы можете привязать шаг к определённой политике повторных попыток, передав WorkflowStepConfig объект в config параметр step.do декоратор. В Python Workflows необходимо убедиться, что ваш dict учитывает WorkflowStepConfig тип.

from workers import WorkflowEntrypoint

class DemoWorkflowClass(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do('step-name', config={"retries": {"limit": 1, "delay": "10 seconds"}})
        async def first_step():
            # do some work
            pass

Если вы определяете ctx параметр, контекст шага внедряется в этот аргумент. Контекст представляет собой словарь со следующими ключами:

from workers import WorkflowEntrypoint

class CtxWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do()
        async def read_context(ctx):
            print(ctx["step"]["name"])    # step name
            print(ctx["step"]["count"])   # step count
            print(ctx["attempt"])         # attempt number
            print(ctx["config"])          # resolved step config
            return ctx["attempt"]

        return await read_context()

Обратите внимание, что env это объект JavaScript, доступный в скрипте Python через JsProxy ↗. Доступ к привязке осуществляется так же, как в JavaScript Worker. См. документация по привязке Workflow чтобы узнать больше о доступных методах.

Рассмотрим предыдущую привязку под названием MY_WORKFLOW. Вот как создать новый экземпляр:

from workers import Response, WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        instance = await self.env.MY_WORKFLOW.create()
        return Response.json({"status": "success"})