Python 处理数据库事务的操作方法

import sqlite3
# 连接到数据库
conn = sqlite3.connect('example.db')
try:
    # 创建游标
    cursor = conn.cursor()
    # 开始事务（默认自动开始，显示调用 conn.commit() 或 conn.rollback() 来结束）
    cursor.execute("BEGIN")
    # 执行多个 SQL 语句
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Alice", 30))
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")
    # 提交事务
    conn.commit()
except Exception as e:
    # 出现错误时回滚事务
    conn.rollback()
    print(f"Transaction failed: {e}")
finally:
    # 关闭连接
    conn.close()

import pymysql
# 连接到 MySQL 数据库
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='password',
    db='test_db'
)
try:
    # 创建游标
    cursor = conn.cursor()
    # 执行多个 SQL 语句
    cursor.execute("INSERT INTO users (name, age) VALUES (%s, %s)", ("Bob", 25))
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 2")
    # 提交事务
    conn.commit()
except Exception as e:
    # 如果有任何错误，回滚事务
    conn.rollback()
    print(f"Transaction failed: {e}")
finally:
    # 关闭连接
    conn.close()

import mysql.connector
# 连接到数据库
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
try:
    cursor = conn.cursor()
    # 执行多条 SQL 语句
    cursor.execute("INSERT INTO orders (order_id, user_id) VALUES (%s, %s)", (101, 1))
    cursor.execute("UPDATE inventory SET quantity = quantity - 1 WHERE product_id = 1")
    # 提交事务
    conn.commit()
except Exception as e:
    # 回滚事务
    conn.rollback()
    print(f"Transaction failed: {e}")
finally:
    conn.close()

import psycopg2
# 连接到 PostgreSQL 数据库
conn = psycopg2.connect(
    dbname="test_db",
    user="user",
    password="password",
    host="localhost"
)
try:
    cursor = conn.cursor()
    # 执行 SQL 操作
    cursor.execute("INSERT INTO employees (name, department) VALUES (%s, %s)", ("John", "HR"))
    cursor.execute("UPDATE payroll SET salary = salary + 500 WHERE employee_id = 3")
    # 提交事务
    conn.commit()
except Exception as e:
    # 事务失败时回滚
    conn.rollback()
    print(f"Transaction failed: {e}")
finally:
    conn.close()

from django.db import transaction
try:
    with transaction.atomic():
        user = User.objects.create(username="alice")
        Account.objects.create(user=user, balance=1000)
except Exception as e:
    print(f"Transaction failed: {e}")

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import User, Account  # 假设已经定义了 ORM 模型
# 连接数据库
engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()
try:
    # 创建用户和账户
    new_user = User(name="Alice")
    session.add(new_user)
    new
_account = Account(user=new_user, balance=1000)
    session.add(new_account)
    # 提交事务
    session.commit()
except Exception as e:
    # 事务回滚
    session.rollback()
    print(f"Transaction failed: {e}")
finally:
    session.close()