2593步骤事务Bfb4e5e9

tools-only2026-10-28查看 889复制 1712

engine.execute("INSERT INTO table VALUES (?)", data) ``` **正确（使用事务）：** ```python from sqlalchemy import text @DBOS.transaction() def save_to_db(name: str, value: str) -> None: sql = text("INSERT INTO my_table (name, value) VALUES (:name, :value)") DBOS.sql_session.execute(sql, {"name": name, "value": value}) @DBOS.transaction() def get_from_db(name: str) -> str | None: sql = text("SELECT value FROM my_table WHERE name = :name LIMIT 1") row = DBOS.sql_session.execute(sql, {"...

Download (.md)

📘如何使用这个技能

1
复制技能内容·点击上方「Copy SKILL.md」复制全部内容，或 Download 下载到本地
2
打开你的 AI 客户端·支持 Claude Code、Trae、Cursor、ChatGPT、DeepSeek、Kimi 等任意 AI 助手
3
粘贴为系统提示/知识文件·Claude Code：放入 .claude/skills/ 目录；普通对话：作为首条消息发送并说「请严格按这份技能文档执行」
4
直接提出你的需求·AI 会按技能定义的工作流、格式和约束来处理，无需再解释背景

💡 提示：技能生效后，每次对话只需一句需求，AI 会按 SKILL.md 的流程执行。

SKILL.md1741 chars

---
title: Use Transactions for Database Operations
impact: HIGH
impactDescription: Transactions provide atomic database operations
tags: transaction, database, postgres, sqlalchemy
---

## Use Transactions for Database Operations

Transactions are a special type of step optimized for database access. They execute as a single database transaction. Only use with Postgres.

**Incorrect (database access in regular step):**

```python
@DBOS.step()
def save_to_db(data):
    # For Postgres, use transactions instead of steps
    # This doesn't get transaction guarantees
    engine.execute("INSERT INTO table VALUES (?)", data)
```

**Correct (using transaction):**

```python
from sqlalchemy import text

@DBOS.transaction()
def save_to_db(name: str, value: str) -> None:
    sql = text("INSERT INTO my_table (name, value) VALUES (:name, :value)")
    DBOS.sql_session.execute(sql, {"name": name, "value": value})

@DBOS.transaction()
def get_from_db(name: str) -> str | None:
    sql = text("SELECT value FROM my_table WHERE name = :name LIMIT 1")
    row = DBOS.sql_session.execute(sql, {"name": name}).first()
    return row[0] if row else None
```

With SQLAlchemy ORM:

```python
from sqlalchemy import Table, Column, String, MetaData, select

greetings = Table("greetings", MetaData(),
    Column("name", String),
    Column("note", String))

@DBOS.transaction()
def insert_greeting(name: str, note: str) -> None:
    DBOS.sql_session.execute(greetings.insert().values(name=name, note=note))
```

Important:
- Only use transactions with Postgres databases
- For other databases, use regular steps
- Never use `async def` with transactions

Reference: [DBOS Transactions](https://docs.dbos.dev/python/reference/decorators#transactions)

来源：skillsdirectory

2593步骤事务Bfb4e5e9

📘如何使用这个技能

评论0