INTEGRITY Dokumentace

← Cloudflare Workflows / workflows / python

Interakce s Workflow

Platforma Python Workers využívá FFI ↗ pro přístup k bindingům na prostředky Cloudflare. Podrobnosti najdete v vazby dokumentaci, kde najdete další informace.

Z hlediska konfigurace vyžaduje povolení Python Workflows přidání python_workflows příznak kompatibility do souboru konfigurace 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"

A takto payload použijete ve svém 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 vazba vám dává přístup k Workflow třída. Všechny její metody jsou dostupné na vazbě.

create

Vytvořte (spusťte) novou instanci daného 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 metoda vrací WorkflowInstance objekt, který lze použít k dotazování na stav instance Workflow. Upozorňujeme, že jde o objekt Javascript, nikoli o objekt Python.

create_batch

Vytvořte (spusťte) dávku nových instancí workflow, až 100 instancí najednou. To se hodí, pokud potřebujete vytvořit více instancí najednou v rámci limit vytváření instancí.

Každý prvek batch se očekává, že tento seznam bude obsahovat jak id a params vlastnosti:

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

Získat instanci workflow podle ID.

Vrátí WorkflowInstance objekt, který lze použít k dotazování na stav instance 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

Odešlete událost do instance 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)

Viz Dokumentace Workflows REST API.

Příkazová řádka (CLI)

Viz Rychlý start s CLI kde se dozvíte více o tom, jak spravovat a spouštět Workflows z příkazové řádky.