Python数据API乐观锁怎么用

wen python案例 2026-07-05 209

本文目录导读：

Python数据API乐观锁怎么用

基于版本号的乐观锁实现
基于时间戳的乐观锁
FastAPI + 乐观锁示例
使用Redis实现乐观锁
最佳实践
注意事项

在Python数据API中实现乐观锁,常见的方法是使用版本号（version）或时间戳（timestamp）机制，以下是具体实现方式：

基于版本号的乐观锁实现

数据库模型示例 (SQLAlchemy)

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
import datetime
Base = declarative_base()
class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    price = Column(Integer)
    stock = Column(Integer)
    version = Column(Integer, default=1)  # 版本号字段
    updated_at = Column(DateTime, default=datetime.datetime.utcnow)

更新操作实现乐观锁

from sqlalchemy.orm import Session
from sqlalchemy import update
def update_product_with_optimistic_lock(session: Session, product_id: int, 
                                       new_price: int, expected_version: int):
    """
    使用乐观锁更新商品价格
    """
    # 查询当前记录
    product = session.query(Product).filter_by(id=product_id).first()
    if not product:
        return {"success": False, "error": "产品不存在"}
    # 检查版本号
    if product.version != expected_version:
        return {"success": False, "error": "数据已被修改，请刷新后重试"}
    # 执行更新（版本号+1）
    stmt = (
        update(Product)
        .where(Product.id == product_id)
        .where(Product.version == expected_version)  # 关键条件
        .values(
            price=new_price,
            version=Product.version + 1,
            updated_at=datetime.datetime.utcnow()
        )
    )
    result = session.execute(stmt)
    session.commit()
    if result.rowcount == 0:
        return {"success": False, "error": "乐观锁冲突，更新失败"}
    return {"success": True, "message": "更新成功"}

基于时间戳的乐观锁

from datetime import datetime
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    balance = Column(Integer)
    lock_version = Column(String(32))  # 存储时间戳的哈希值
    def before_update(self):
        self.lock_version = str(datetime.utcnow().timestamp())
# 更新函数
def update_user_balance(session: Session, user_id: int, 
                        new_balance: int, current_lock_version: str):
    user = session.query(User).filter_by(id=user_id).first()
    if user.lock_version != current_lock_version:
        return {"success": False, "error": "数据已被修改"}
    user.balance = new_balance
    user.before_update()  # 更新锁版本
    session.commit()
    return {"success": True}

FastAPI + 乐观锁示例

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
app = FastAPI()
class ProductUpdate(BaseModel):
    id: int
    name: str
    price: float
    version: int  # 客户端传递当前版本号
@app.put("/products/{product_id}")
async def update_product(
    product_id: int,
    product_update: ProductUpdate,
    db: Session = Depends(get_db)
):
    # 使用原生SQL实现乐观锁
    from sqlalchemy import text
    sql = text("""
        UPDATE products 
        SET name = :name, 
            price = :price, 
            version = version + 1,
            updated_at = NOW()
        WHERE id = :id 
        AND version = :expected_version
    """)
    result = db.execute(sql, {
        "id": product_id,
        "name": product_update.name,
        "price": product_update.price,
        "expected_version": product_update.version
    })
    db.commit()
    if result.rowcount == 0:
        raise HTTPException(
            status_code=409, 
            detail="数据已被其他用户修改，请刷新后重试"
        )
    return {"message": "更新成功"}

使用Redis实现乐观锁

import redis
import json
r = redis.Redis(host='localhost', port=6379, db=0)
def update_with_redis_optimistic_lock(key: str, update_func, retry=3):
    """
    使用Redis WATCH命令实现乐观锁
    """
    with r.pipeline() as pipe:
        while True:
            try:
                # 监控key
                pipe.watch(key)
                # 读取当前数据
                current_data = json.loads(pipe.get(key))
                # 执行更新逻辑
                new_data = update_func(current_data)
                # 开始事务
                pipe.multi()
                pipe.set(key, json.dumps(new_data))
                pipe.execute()
                return True
            except redis.WatchError:
                # 数据被修改，重试
                retry -= 1
                if retry <= 0:
                    raise Exception("更新失败，多次重试后仍冲突")
                continue

最佳实践

数据库层实现（推荐）

-- 在数据库层面实现乐观锁
CREATE TABLE products (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    price DECIMAL(10,2),
    stock INT,
    version INT DEFAULT 1,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
-- 更新时检查版本号
UPDATE products 
SET price = 100, 
    version = version + 1 
WHERE id = 1 
AND version = 1;  -- 期望的版本号

客户端代码

class ProductService:
    def update_product(self, product_id, new_price, expected_version, max_retries=3):
        retry_count = 0
        while retry_count < max_retries:
            result = self.db.execute(...)
            if result.rowcount > 0:
                return {"success": True}
            retry_count += 1
            # 获取最新版本号
            product = self.get_product(product_id)
            expected_version = product.version
        return {"success": False, "error": "更新失败，请重试"}