Python依赖适配案例实操？

wen python案例 2026-06-08 06:09:25 2

本文目录导读：

案例场景
问题诊断
适配方案
代码适配实操
版本冲突解决方案
自动化适配脚本
测试验证
最佳实践建议

我来通过一个实际案例,演示Python依赖适配的完整流程。

案例场景

假设需要将一个使用 FastAPI 0.68.0 + SQLAlchemy 1.3 + Pydantic 1.8 的项目升级到最新版本。

问题诊断

原始依赖文件 (requirements.txt)

fastapi==0.68.0
uvicorn==0.15.0
sqlalchemy==1.3.23
pydantic==1.8.2
alembic==1.7.5
python-jose==3.3.0
passlib==1.7.4
python-multipart==0.0.5

检查依赖冲突

# 使用 pipdeptree 查看依赖树
pip install pipdeptree
pipdeptree
# 输出示例：
fastapi==0.68.0
  - pydantic [required: >=1.6.2,<2.0.0, installed: 1.8.2]
  - starlette [required: ==0.14.2, installed: 0.14.2]
sqlalchemy==1.3.23
  - greenlet [required: !=0.4.17, installed: 1.1.2]

适配方案

渐进式升级（推荐）

# new_requirements.txt
fastapi==0.100.0        # 升级到支持异步
uvicorn[standard]==0.23.0
sqlalchemy==2.0.20      # 重大版本变更
pydantic==2.1.4         # v2 版本
alembic==1.12.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6

保持兼容性升级

# compatible_requirements.txt
fastapi>=0.68.0,<0.90.0
sqlalchemy>=1.4.0,<2.0.0  # 使用1.4过渡版本
pydantic>=1.9.0,<2.0.0   # 保持在v1.x

代码适配实操

1 Pydantic v1 → v2 迁移

原始代码 (v1):

from pydantic import BaseModel, validator
class UserIn(BaseModel):
    username: str
    password: str
    @validator('password')
    def validate_password(cls, v):
        if len(v) < 8:
            raise ValueError('密码至少8位')
        return v
    class Config:
        orm_mode = True

适配后代码 (v2):

from pydantic import BaseModel, field_validator
from pydantic.config import ConfigDict
class UserIn(BaseModel):
    username: str
    password: str
    @field_validator('password')
    @classmethod
    def validate_password(cls, v):
        if len(v) < 8:
            raise ValueError('密码至少8位')
        return v
    model_config = ConfigDict(from_attributes=True)  # 替代 orm_mode

2 SQLAlchemy 1.3 → 2.0 迁移

原始代码 (1.3):

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///./test.db')
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
# 查询
session.query(User).filter(User.name == 'test').first()

适配后代码 (2.0):

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from typing import Optional
class Base(DeclarativeBase):
    pass
class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[Optional[str]] = mapped_column(String(50))
# 使用新式查询
with Session(engine) as session:
    user = session.execute(
        select(User).where(User.name == 'test')
    ).scalar_one_or_none()
# 或使用新式Mapped方法
user = await session.get(User, 1)

3 FastAPI 依赖注入适配

原始代码:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
app = FastAPI()
def get_db():
    db = Session()
    try:
        yield db
    finally:
        db.close()
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

适配后代码 (支持异步):

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
engine = create_async_engine("sqlite+aiosqlite:///./test.db")
async def get_db():
    async with AsyncSession(engine) as session:
        yield session
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

版本冲突解决方案

使用 pip 工具解决冲突

# 分析冲突
pip install pip-tools
pip-compile requirements.in
# 使用 pipdeptree 可视化
pipdeptree --warn silence | grep -E "^\w+"
# 强制安装特定版本
pip install package==version --no-deps

使用 virtualenv 隔离环境

# 创建多环境测试
python -m venv env_v1
python -m venv env_v2
# 在v1环境测试旧代码
source env_v1/bin/activate
pip install -r old_requirements.txt
python test_old.py
# 在v2环境测试新代码
source env_v2/bin/activate
pip install -r new_requirements.txt
python test_new.py

自动化适配脚本

# update_dependencies.py
import subprocess
import json
def check_dependencies():
    """检查当前依赖状态"""
    result = subprocess.run(
        ['pip', 'list', '--format=json'],
        capture_output=True, text=True
    )
    return json.loads(result.stdout)
def resolve_conflicts(packages):
    """自动解决已知冲突"""
    conflicts = {
        'pydantic': {
            'old': '<2.0.0',
            'new': '>=2.0.0',
            'changes': {
                'validator': 'field_validator',
                'Config': 'model_config'
            }
        },
        'sqlalchemy': {
            'old': '<2.0.0',
            'new': '>=2.0.0',
            'changes': {
                'declarative_base': 'DeclarativeBase',
                'Column': 'mapped_column'
            }
        }
    }
    return conflicts
def auto_update():
    """自动更新依赖"""
    deps = check_dependencies()
    conflicts = resolve_conflicts(deps)
    for pkg, info in conflicts.items():
        print(f"处理 {pkg} 冲突...")
        print(f"  旧版本: {info['old']}")
        print(f"  新版本: {info['new']}")
        print(f"  需要修改: {info['changes']}")
if __name__ == "__main__":
    auto_update()

测试验证

# test_adaptation.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
def test_api_compatibility():
    """测试API兼容性"""
    client = TestClient(app)
    # 测试新接口
    response = client.get("/users/1")
    assert response.status_code == 200
    # 测试Pydantic验证
    response = client.post("/users/", json={
        "username": "test",
        "password": "123"  # 应该失败
    })
    assert response.status_code == 422
def test_db_compatibility():
    """测试数据库兼容性"""
    # 测试新旧查询
    from app.models import User
    assert hasattr(User, 'name')  # 字段存在