Python数据API数据完整性怎么保证

wen python案例 2026-07-05 91

本文目录导读：

传输层完整性
数据验证层
并发控制
数据库约束
审计日志
数据一致性模式
监控与告警
完整示例：REST API 实现

在 Python 数据 API 中保证数据完整性，主要涉及传输、存储、并发与校验四个层面，以下是系统性的实践方法：

传输层完整性

HTTPS/TLS 加密

# 强制使用 HTTPS
from flask import Flask, request, abort
import ssl
app = Flask(__name__)
@app.before_request
def enforce_https():
    if request.headers.get('X-Forwarded-Proto', 'http') == 'http':
        abort(403, description="HTTPS required")

数据校验和

import hashlib
import hmac
def generate_checksum(data, secret_key):
    """生成数据校验和"""
    return hmac.new(
        secret_key.encode(), 
        data.encode(), 
        hashlib.sha256
    ).hexdigest()
def verify_checksum(data, checksum, secret_key):
    """验证数据完整性"""
    expected = generate_checksum(data, secret_key)
    return hmac.compare_digest(expected, checksum)

数据验证层

Pydantic 模型验证

from pydantic import BaseModel, ValidationError, Field
from datetime import datetime
from typing import Optional
class UserData(BaseModel):
    id: int = Field(gt=0)
    name: str = Field(min_length=1, max_length=100)
    email: str = Field(pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    created_at: datetime
    version: Optional[int] = 1
# 自动验证输入
def create_user(user_data: dict):
    try:
        validated = UserData(**user_data)
        return validated.dict()
    except ValidationError as e:
        raise APIError(f"Invalid data: {e}")

自定义验证器

from pydantic import validator
class Transaction(BaseModel):
    amount: float
    currency: str
    timestamp: datetime
    @validator('amount')
    def validate_positive(cls, v):
        if v <= 0:
            raise ValueError('Amount must be positive')
        return v
    @validator('currency')
    def validate_currency(cls, v):
        valid_currencies = ['USD', 'EUR', 'CNY']
        if v not in valid_currencies:
            raise ValueError(f'Invalid currency: {v}')
        return v

并发控制

乐观锁

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Account(Base):
    __tablename__ = 'accounts'
    id = Column(Integer, primary_key=True)
    balance = Column(Integer, default=0)
    version = Column(Integer, default=1)  # 版本号
    def update_balance(self, amount, expected_version):
        from sqlalchemy import update
        stmt = update(Account).where(
            Account.id == self.id,
            Account.version == expected_version
        ).values(
            balance=Account.balance + amount,
            version=Account.version + 1
        )
        result = session.execute(stmt)
        if result.rowcount == 0:
            raise ConcurrentUpdateError("Data was modified by another process")

悲观锁（行级锁）

from sqlalchemy import select
from sqlalchemy.orm import with_for_update
def transfer_money(from_account_id, to_account_id, amount):
    # 使用悲观锁锁定行
    from_account = session.query(Account).filter(
        Account.id == from_account_id
    ).with_for_update().first()
    if from_account.balance < amount:
        raise InsufficientBalanceError()
    to_account = session.query(Account).filter(
        Account.id == to_account_id
    ).with_for_update().first()
    from_account.balance -= amount
    to_account.balance += amount
    session.commit()

数据库约束

SQLAlchemy 模型约束

class Product(Base):
    __tablename__ = 'products'
    __table_args__ = (
        UniqueConstraint('sku', name='unique_sku'),
        CheckConstraint('price > 0', name='positive_price'),
        CheckConstraint('stock >= 0', name='non_negative_stock'),
    )
    id = Column(Integer, primary_key=True)
    sku = Column(String(50), nullable=False)
    price = Column(Numeric(10,2), nullable=False)
    stock = Column(Integer, default=0)

审计日志

import logging
from datetime import datetime
from functools import wraps
def audit_log(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        audit_entry = {
            'timestamp': datetime.utcnow(),
            'function': func.__name__,
            'args': args,
            'kwargs': kwargs,
            'result': result
        }
        # 保存审计日志
        save_audit_log(audit_entry)
        return result
    return wrapper
class AuditService:
    @staticmethod
    def save_audit_log(entry):
        # 写入审计数据库或日志文件
        logging.info(f"Audit: {entry}")
        # 或保存到数据库
        with session_scope() as session:
            audit = AuditLog(**entry)
            session.add(audit)

数据一致性模式

事务管理

from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
@contextmanager
def transaction_session():
    """事务管理上下文管理器"""
    session = Session()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()
# 使用示例
with transaction_session() as session:
    order = Order(user_id=1, total=100)
    session.add(order)
    # 如果这里失败，整个事务回滚
    payment = Payment(order_id=order.id, amount=100)
    session.add(payment)

重试机制（幂等性）

import time
from functools import wraps
def retry_on_conflict(max_retries=3, base_delay=0.1):
    """自动重试机制"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (ConcurrentUpdateError, DeadlockError) as e:
                    last_exception = e
                    delay = base_delay * (2 ** attempt)
                    time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator
@retry_on_conflict(max_retries=3)
def update_inventory(product_id, quantity):
    # 更新库存，遇到冲突自动重试
    pass

监控与告警

from prometheus_client import Counter, Histogram
import time
DATA_INTEGRITY_ERRORS = Counter(
    'data_integrity_errors_total',
    'Total data integrity errors',
    ['error_type']
)
VALIDATION_ERRORS = Counter(
    'validation_errors_total',
    'Total validation errors',
    ['field']
)
def monitor_integrity(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            return result
        except ValidationError as e:
            VALIDATION_ERRORS.labels(
                error_type='validation'
            ).inc()
            raise
        except Exception as e:
            DATA_INTEGRITY_ERRORS.labels(
                error_type='unknown'
            ).inc()
            raise
        finally:
            duration = time.time() - start
            # 记录处理时间
    return wrapper

完整示例：REST API 实现

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
app = FastAPI()
class OrderCreate(BaseModel):
    user_id: int
    items: list[dict]
    total: float
    class Config:
        # 禁止额外字段
        extra = 'forbid'
@app.post("/orders")
async def create_order(order: OrderCreate):
    with transaction_session() as session:
        try:
            # 1. 验证订单数据
            validate_order_items(order.items)
            # 2. 检查用户存在
            user = session.query(User).filter(
                User.id == order.user_id
            ).with_for_update().first()
            if not user:
                raise HTTPException(status_code=404, detail="User not found")
            # 3. 创建订单（使用版本控制）
            new_order = Order(
                user_id=order.user_id,
                total=order.total,
                status='pending',
                version=1
            )
            session.add(new_order)
            session.flush()
            # 4. 更新库存
            for item in order.items:
                update_inventory(item['product_id'], -item['quantity'])
            # 5. 提交事务
            session.commit()
            return {"order_id": new_order.id, "status": "success"}
        except Exception as e:
            session.rollback()
            raise HTTPException(status_code=500, detail=str(e))