发现优质的 AI Agent 技能
聚合 Claude Skills、LangChain、AutoGPT 等优质资源,助力开发者快速构建智能应用
Wrong: assuming the workflow stopped immediately
cleanup_resources() # May race with workflow still running its current step ``` **Correct (wait for cancellation to complete):**
Never start workflows from inside a step!
DBOS.start_workflow(another_workflow) ``` **Incorrect (modifying global state):**
DBOS state from previous test!
result = another_workflow() ``` **Correct (reset fixture):**
External API call directly in workflow - not checkpointed!
response = requests.get("https://api.example.com/data") return response.json() ``` **Correct (external call in step):**
Multiple requests = multiple workflows for same user!
queue.enqueue(process_workflow, user_id) ``` **Correct (deduplicated by user):**
Starting many workflows without control
for task in tasks: DBOS.start_workflow(process_task, task) # Could overwhelm resources ``` **Correct (using queue):**
Don't use external cron or manual timers
import schedule schedule.every(1).minute.do(my_task) ``` **Correct (DBOS scheduled workflow):**
Multiple clicks = multiple payments!
handle = DBOS.start_workflow(payment_workflow, order_id) return handle.get_result() ``` **Correct (idempotent with workflow ID):**
Expensive processing
analyze(user_input) @app.post("/input") def on_input(user_id: str, input: str): DBOS.start_workflow(process_input, input) ```
instance_name must be unique and passed to super()
super().__init__(instance_name=url) @DBOS.workflow() def fetch_workflow(self): return self.fetch_url()
Don't configure at module level!
config: DBOSConfig = {"name": "my-app"} DBOS(config=config) @app.get("/") @DBOS.workflow() def endpoint():
Don't configure at module level!
config: DBOSConfig = { "name": "my-app", } DBOS(config=config) @DBOS.workflow()