INTEGRITY Документация

← Cloudflare Workflows / workflows / python

Взаимодействие с Workflow

Платформа Python Workers использует FFI ↗ для доступа к привязкам ресурсов Cloudflare. См. привязки документации.

С точки зрения конфигурации для включения Python Workflows потребуется добавить python_workflows флаг совместимости в файл конфигурации Wrangler.

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "workflows-starter",
	"main": "src/index.py",
	// Set this to today's date
	"compatibility_date": "2026-08-28",
	"compatibility_flags": ["python_workflows", "python_workers"],
	"workflows": [
		{
			// name of your workflow
			"name": "workflows-starter",
			// binding name env.MY_WORKFLOW
			"binding": "MY_WORKFLOW",
			// this is class that extends the Workflow class in src/index.py
			"class_name": "MyWorkflow",
		}
	]
}
"$schema" = "./node_modules/wrangler/config-schema.json"
name = "workflows-starter"
main = "src/index.py"
# Set this to today's date
compatibility_date = "2026-08-28"
compatibility_flags = [ "python_workflows", "python_workers" ]

[[workflows]]
name = "workflows-starter"
binding = "MY_WORKFLOW"
class_name = "MyWorkflow"

А вот как использовать payload в своем Workflow:

from workers import WorkflowEntrypoint

class DemoWorkflowClass(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do('step-name')
        async def first_step():
            payload = event["payload"]
            return payload

Workflow

Workflow привязка предоставляет доступ к Workflow класс. Все его методы доступны в привязке.

create

Создайте (запустите) новый экземпляр заданного Workflow.

from workers import WorkerEntrypoint, Response


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

create метод возвращает WorkflowInstance объект, с помощью которого можно запросить статус экземпляра workflow. Обратите внимание, что это объект Javascript, а не объект Python.

create_batch

Создайте (запустите) пакет новых экземпляров Workflow, до 100 экземпляров за раз. Это полезно, если вам нужно создать несколько экземпляров одновременно в пределах лимит создания экземпляров.

Каждый элемент массива batch список должен включать как id и params свойства:

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
			# Create a new batch of 3 Workflow instances, each with its own ID and pass params to the Workflow instances
        instances = [
            {"id": "id-abc123", "params": {"hello": "world-0"}},
            {"id": "id-def456", "params": {"hello": "world-1"}},
            {"id": "id-ghi789", "params": {"hello": "world-2"}},
        ]
        await self.env.MY_WORKFLOW.create_batch(instances)
        return Response.json({"status": "success"})

get

Получить экземпляр workflow по ID.

Возвращает WorkflowInstance объект, с помощью которого можно запросить статус экземпляра workflow.

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        instance = await self.env.MY_WORKFLOW.get("abc-123")

        # FFI methods available for WorkflowInstance
        await instance.status()
        await instance.pause()
        await instance.resume()
        await instance.restart()
        await instance.terminate()
        return Response.json({"status": "success"})

send_event

Отправьте событие в экземпляр workflow.

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        await self.env.MY_WORKFLOW.send_event(type="my-event-type", payload={"foo": "bar"})
        return Response.json({"status": "success"})

REST API (HTTP)

См. Документация по Workflows REST API.

Командная строка (CLI)

См. Быстрый старт CLI чтобы узнать больше об управлении и запуске Workflows через командную строку.