SQLAlchemy 2.0 与 Pydantic 模型类型安全集成指南

from datetime import datetime
from typing import Optional

from pydantic import BaseModel, Field, EmailStr, ConfigDict
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import declarative_base

# SQLAlchemy Base
Base = declarative_base()

# Pydantic 模型
class UserPydantic(BaseModel):
    model_config = ConfigDict(from_attributes=True) # Pydantic v2
    # orm_mode = True # Pydantic v1
    name: str = Field(...)
    email: EmailStr()
    is_active: bool = Field(default=True)
    is_admin: bool = Field(default=False)
    created_at: datetime = Field(...)

# SQLAlchemy ORM 模型 (1.x 风格)
class UserDB(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String, index=True)
    email = Column(String, index=True, nullable=False, unique=True)
    hashed_password = Column(String(length=255), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    is_admin = Column(Boolean, default=False, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

    def to_pydantic(self) -> UserPydantic:
        return UserPydantic(
            name=self.name,
            email=self.email,
            is_active=self.is_active,
            is_admin=self.is_admin,
            created_at=self.created_at
        )

from datetime import datetime
from typing import Optional, List
from sqlalchemy.orm import Mapped, relationship, mapped_column
from sqlalchemy import String, Boolean, DateTime, Integer, func

# 假设 Base 已经定义，如 from sqlalchemy.orm import DeclarativeBase; class Base(DeclarativeBase): pass
# 为了演示，我们使用一个简单的 Base
from sqlalchemy.orm import declarative_base
Base = declarative_base()

# Pydantic 模型保持不变
class UserPydantic(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    name: str = Field(...)
    email: EmailStr()
    is_active: bool = Field(default=True)
    is_admin: bool = Field(default=False)
    created_at: datetime = Field(...)
    # 如果 Pydantic 模型也需要包含关系，可以这样定义：
    # instagram_dms: List["InstagramDmPydantic"] = [] # 假设 InstagramDmPydantic 存在

# 假设存在一个 InstagramDmDB 模型用于演示关系
class InstagramDmDB(Base):
    __tablename__ = "instagram_dms"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    message: Mapped[str] = mapped_column(String)
    user_id: Mapped[int] = mapped_column(Integer, index=True)
    user: Mapped["UserDB"] = relationship("UserDB", back_populates="instagram_dms")

# SQLAlchemy ORM 模型 (2.0 风格)
class UserDB(Base):
    __tablename__ = "users"
    # 使用 Mapped 和 mapped_column 声明属性
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String, index=True)
    email: Mapped[str] = mapped_column(String, index=True, nullable=False, unique=True)
    hashed_password: Mapped[str] = mapped_column(String(length=255), nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    is_admin: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=func.now(), nullable=False) # 使用 func.now() 获取数据库时间

    # 声明关系，同样使用 Mapped
    instagram_dms: Mapped[List["InstagramDmDB"]] = relationship("InstagramDmDB", back_populates="user")

    # 现在不再需要 to_pydantic 方法，Pydantic 可以直接从 ORM 实例创建
    # def to_pydantic(self) -> UserPydantic:
    #     # 这段代码现在是多余的，但如果需要手动映射，类型检查器会认为 self.name 是 str
    #     return UserPydantic(
    #         name=self.name,
    #         email=self.email,
    #         is_active=self.is_active,
    #         is_admin=self.is_admin,
    #         created_at=self.created_at
    #     )

# 假设我们已经从数据库中获取了一个 UserDB 实例
# from sqlalchemy import create_engine
# from sqlalchemy.orm import sessionmaker
#
# engine = create_engine("sqlite:///:memory:")
# Base.metadata.create_all(engine)
# Session = sessionmaker(bind=engine)
# session = Session()
#
# new_user = UserDB(
#     name="Alice",
#     email="alice@example.com",
#     hashed_password="hashed_password_abc",
#     is_active=True,
#     is_admin=False,
#     created_at=datetime.utcnow()
# )
# session.add(new_user)
# session.commit()
# session.refresh(new_user)
#
# user_from_db = session.query(UserDB).first()

# 模拟一个从数据库获取的 UserDB 实例
class MockUserDB:
    def __init__(self):
        self.name = "Test User"
        self.email = "test@example.com"
        self.is_active = True
        self.is_admin = False
        self.created_at = datetime.utcnow()
        self.id = 1 # Pydantic from_attributes 不会使用 id 除非 Pydantic 模型也定义了 id

mock_user_instance = MockUserDB()

# 使用 Pydantic 的 from_attributes 功能直接从 ORM 实例创建 Pydantic 模型
user_pydantic_instance = UserPydantic.model_validate(mock_user_instance) # Pydantic v2
# user_pydantic_instance = UserPydantic.from_orm(mock_user_instance) # Pydantic v1

print(user_pydantic_instance.model_dump_json(indent=2))

{
  "name": "Test User",
  "email": "test@example.com",
  "is_active": true,
  "is_admin": false,
  "created_at": "2023-10-27T10:00:00.000000"
}