← Cloudflare Workflows / workflows / python

Python Workers API

Tento návod se věnuje Python Workflows SDK a vysvětluje, jak pomocí Pythonu vytvářet workflows.

WorkflowEntrypoint je hlavním vstupním bodem pro Python workflow. Rozšiřuje WorkflowEntrypoint třídu, a implementuje run metoda.

from workers import WorkflowEntrypoint

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

• step.do(name=None, *, concurrent=False, config=None), dekorátor, který umožňuje definovat krok ve workflow.

  • name, volitelný název kroku. Pokud jej vynecháte, použije se název funkce (func.__name__) se použije.
  • concurrent, volitelná booleovská hodnota, která určuje, zda mohou závislosti tohoto kroku běžet souběžně.
  • config, volitelný WorkflowStepConfig pro konfiguraci chování opakování specifické pro krok. To se předává jako slovník Pythonu a poté je typově převedeno na WorkflowStepConfig objekt.

  Všechny parametry kromě name lze zadat pouze jako klíčové argumenty.

Závislosti se řeší implicitně podle názvu parametru. Pokud název parametru funkce kroku odpovídá dříve deklarované funkci kroku, do daného kroku se vloží její výsledek.

Pokud definujete ctx parametr, do daného argumentu se vloží kontext kroku.

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()

Upozorňujeme, že dekorátor sám o sobě krok nevolá, pouze vrací volatelný objekt, kterým lze krok vyvolat. Aby se krok spustil, musíte tento volatelný objekt zavolat.

Při vrácení stavu z kroku musíte zajistit, aby byla vrácená hodnota serializovatelná.

• step.sleep(name, duration)
  • name, název kroku.
  • duration, doba, po kterou se má čekat, zadaná jako number v milisekundách nebo jako WorkflowDuration-kompatibilní řetězec.

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

• step.sleep_until(name, timestamp)
  • name, název kroku.
  • timestamp : datetime.datetime objekt nebo počet sekund od unixové epochy, do kterého se má instance Workflow uspat.

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, název kroku.
  • event_type, typ události, na kterou se čeká.
  • timeout, časový limit pro wait_for_event volání. Výchozí timeout je 24 hodin.

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

event parametr je slovník, který obsahuje payload předaný instanci Workflow, spolu s dalšími metadaty:

• payload - payload předaný instanci Workflow.
• timestamp - časové razítko spuštění Workflow.
• instanceId - ID aktuální instance Workflow.
• workflowName - název Workflow.

Sémantika Workflows umožňuje uživatelům zachytávat výjimky vyvolané až na nejvyšší úrovni.

Zachytávání konkrétních výjimek uvnitř except blok nemusí fungovat, protože některé chyby Pythonu nebudou při průchodu vrstvou RPC znovu vytvořeny jako stejný typ chyby.

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 poskytuje NonRetryableError třída, kterou lze použít k signalizaci, že se krok nemá opakovat.

from workers.workflows import NonRetryableError

raise NonRetryableError(message)

Krok můžete svázat s konkrétní zásadou opakování předáním WorkflowStepConfig objekt do config parametru step.do dekorátor. U Python Workflows musíte zajistit, aby váš dict respektuje WorkflowStepConfig typ.

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

Pokud definujete ctx parametr, kontext kroku je vloženo do tohoto argumentu. Kontext je slovník s následujícími klíči:

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()

Upozorňujeme, že env je JavaScript objekt, který je Python skriptu vystaven přes JsProxy ↗. K vazbě můžete přistupovat stejně jako u JavaScript Workeru. Podívejte se na Dokumentace k vazbě na Workflow kde se dozvíte více o dostupných metodách.

Uvažujme předchozí binding s názvem MY_WORKFLOW. Takto byste vytvořili novou instanci:

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"})