发现优质的 AI Agent 技能
聚合 Claude Skills、LangChain、AutoGPT 等优质资源,助力开发者快速构建智能应用
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()
Client code to read events
@app.post("/start") def start_workflow(): handle = DBOS.start_workflow(long_workflow) return {"workflow_id": handle.get_workflow_id()} @app.get("/status/{workflow_id}")
Missing workflow_name and queue_name!
handle = client.enqueue({}, task_data) ``` **Correct (with required options):**
Deploying new code directly kills in-progress workflows
@DBOS.workflow() def workflow(): step_a() step_b()
Original
@DBOS.workflow() def workflow(): foo() bar() @DBOS.workflow()
Don't use sync sleep in async workflow!
DBOS.sleep(10) handle = DBOS.start_workflow(other_async_workflow) ```
workflow timeout
来自 sickn33/antigravity-awesome-skills 技能