AI原生开发:ClaudeCode与国产大模型协同实战 第5章:给野马套上缰绳——用OpenSpec把变成死磕契约的偏执狂

# point_deduct_spec.yamlopen_spec_version: "1.0"module: PointsDeductionentities:  PointAccount:    description: "用户积分账户"    fields:      user_id: { type: string, required: true }      balance: { type: integer, required: true, minimum: 0, description: "可用余额，绝对不允许为负" }      version: { type: integer, required: true, description: "乐观锁版本号" }  PointTransactionLog:    description: "积分流水记录"    fields:      log_id: { type: string, required: true }      user_id: { type: string, required: true }      amount: { type: integer, required: true, description: "变动正负数" }      reason: { type: string, required: true }states:  PointAccount:    - ACTIVE    - FROZENtransitions:  deduct_points:    description: "扣减积分核心流转"    from_state: ACTIVE    to_state: ACTIVE  # 扣减不改变状态，除非余额归零    pre_conditions:      - "account.balance >= deduct_amount"      - "deduct_amount > 0"    post_conditions:      - "account.balance >= 0" # 铁律：绝对不能超卖    side_effects:      - "写入 PointTransactionLog"contracts:  DeductPointsAPI:    endpoint: /points/deduct    method: POST    request:      user_id: { type: string, required: true }      amount: { type: integer, required: true, minimum: 1, description: "扣减数量，必须大于0" }      order_id: { type: string, required: true, description: "幂等键，同一订单号只扣一次" }    response:      success:        code: 200        data: { remaining_balance: integer }      errors:        - code: 40001          message: "Insufficient balance"        - code: 40002          message: "Duplicate deduction (Idempotent hit)"        - code: 40003          message: "Account frozen"

# Role你是一个极度严谨的支付结算后端工程师。# Task请根据我提供的 OpenSpec 规约文档，实现 `deduct_points` 的核心业务逻辑。# OpenSpec 规约（将上面 point_deduct_spec.yaml 的内容完整粘贴在这里）# Directive1. 必须使用 Python + SQLAlchemy (同步模式)。2. 必须严格遵循 OpenSpec 中的 `transitions` 和 `contracts` 定义。3. 针对 `pre_conditions`，必须使用数据库层面的乐观锁（基于 version 字段）来保证并发安全，绝对不能出现超卖（余额为负）。4. 必须实现基于 order_id 的幂等逻辑。5. 异常必须严格按照 contracts 中定义的 errors 抛出。请直接输出核心的 service 层代码（包含实体模型和业务逻辑）。

import hashlibfrom sqlalchemy import Column, String, Integer, create_enginefrom sqlalchemy.orm import declarative_base, sessionmakerfrom sqlalchemy.exc import IntegrityErrorBase = declarative_base()# ================== 实体模型 (严格对应 OpenSpec Entities) ==================class PointAccount(Base):    __tablename__ = 'point_account'    user_id = Column(String(64), primary_key=True)    balance = Column(Integer, nullable=False) # OpenSpec: minimum: 0    version = Column(Integer, nullable=False) # OpenSpec: 乐观锁版本号class PointTransactionLog(Base):    __tablename__ = 'point_transaction_log'    log_id = Column(String(128), primary_key=True)    user_id = Column(String(64), nullable=False)    amount = Column(Integer, nullable=False)    reason = Column(String(255), nullable=False)class IdempotentRecord(Base):    __tablename__ = 'idempotent_record'    order_id = Column(String(128), primary_key=True)    user_id = Column(String(64), nullable=False)# ================== 自定义异常 (严格对应 OpenSpec Contracts.errors) ==================class InsufficientBalanceError(Exception):    def __init__(self): self.code = 40001; self.message = "Insufficient balance"class DuplicateDeductionError(Exception):    def __init__(self): self.code = 40002; self.message = "Duplicate deduction (Idempotent hit)"class AccountFrozenError(Exception):    def __init__(self): self.code = 40003; self.message = "Account frozen"# ================== 核心业务逻辑 (严格对应 OpenSpec Transitions) ==================class PointService:    def __init__(self, session):        self.session = session    def deduct_points(self, user_id: str, amount: int, order_id: str):        # 1. 幂等校验 (OpenSpec Contract: order_id 幂等键)        existing_record = self.session.query(IdempotentRecord).filter_by(order_id=order_id).first()        if existing_record:            raise DuplicateDeductionError()        # 2. 基础校验 (OpenSpec Contract: amount minimum 1)        if amount <= 0:            raise ValueError("Deduct amount must be greater than 0")        # 3. 查询账户        account = self.session.query(PointAccount).filter_by(user_id=user_id).with_for_update().first()        if not account:            raise ValueError("Account not found")        # 4. 前置条件校验 (OpenSpec Transitions: pre_conditions)        if account.balance < amount:            raise InsufficientBalanceError()        # 5. 执行扣减与乐观锁更新 (OpenSpec Transitions: post_conditions balance >= 0)        # UPDATE point_account SET balance = balance - :amount, version = version + 1         # WHERE user_id = :user_id AND version = :old_version        updated_rows = self.session.query(PointAccount).filter(            PointAccount.user_id == user_id,            PointAccount.version == account.version,            PointAccount.balance >= amount # 数据库层最后防线        ).update({            "balance": PointAccount.balance - amount,            "version": PointAccount.version + 1        })        if updated_rows == 0:            # 乐观锁冲突或余额不足，重试或抛异常            raise InsufficientBalanceError()        # 6. 记录流水 (OpenSpec Transitions: side_effects)        log_id = hashlib.md5(f"{order_id}:{user_id}".encode()).hexdigest()        new_log = PointTransactionLog(            log_id=log_id,            user_id=user_id,            amount=-amount, # 负数表示扣减            reason=f"Deduction for order {order_id}"        )        self.session.add(new_log)        # 7. 记录幂等键        idempotent_rec = IdempotentRecord(order_id=order_id, user_id=user_id)        self.session.add(idempotent_rec)        self.session.commit()        # 8. 返回 (OpenSpec Contract: response success)        return {"remaining_balance": account.balance - amount}# ================== 测试运行 ==================if __name__ == "__main__":    # 初始化内存 SQLite 数据库测试    engine = create_engine('sqlite:///:memory:')    Base.metadata.create_all(engine)    Session = sessionmaker(bind=engine)    session = Session()    # 准备测试数据    session.add(PointAccount(user_id="U001", balance=100, version=1))    session.commit()    service = PointService(session)    print("--- 正常扣减 ---")    res1 = service.deduct_points(user_id="U001", amount=30, order_id="ORD1001")    print(f"扣减成功，剩余余额: {res1['remaining_balance']}")    print("\n--- 幂等拦截 ---")    try:        service.deduct_points(user_id="U001", amount=30, order_id="ORD1001")    except DuplicateDeductionError as e:        print(f"拦截成功: {e.message} (Code: {e.code})")    print("\n--- 余额不足 ---")    try:        service.deduct_points(user_id="U001", amount=999, order_id="ORD1002")    except InsufficientBalanceError as e:        print(f"拦截成功: {e.message} (Code: {e.code})")