Skip to content

Mapped Class

Mapped Class即映射类，他是 ORM 的核心。在 SQLAlchemy 中提供了两种形式的映射类:

  • 声明式映射(Declarative Mapping): 目前推荐的方式，类似 Python 常规的类定义
  • 命令式映射(Imperative Mapping): 通过命令来包装空类，他主要用于对现有的代码进行包装(内省也通过这种方式来进行类映射)

registry

registry是映射类的通用注册表，他提供了映射类的配置钩子。registry 类似 SQLALchemy Core 中的MetaData来表示描述数据库，实际上创建 registry 对象的第一个参数就是 MetaData:

Python
default_registry = registry()
my_registry = registry(MetaData(schema='cnopendata'))

就像 Core 中所有的 Table 都应该注册到 MetaData 一样，所有的映射类(无论是声明式映射还是命令式映射)也都应该被注册到 registry 对象中，这是通过 registry 对象上绑定的三个方法来实现的:

  • registry.generate_base(): 返回一个声明式映射基类(DeclarativeBase)，所有基于该基类创建的类都会被注册到 registry 中
  • registry.mapped(): 类装饰器，他类似声明式映射基类，只不过使用装饰器的形式来注册
  • registry.map_imperatively(class, Table): 使用 class(空类)对 Table 进行包装，这个是命令式映射的核心方法

由于 MetaData 也包含在 registry 中，因此创建绑定到 registry 中表的方式:

Python
# 全部创建
Base.metadata.create_all(engine)

# 单表创建
Base.metadata.tables['user'].create(engine)

# 默认情况如果存在创建会失败，可以指定 checkfirst = True 来在创建前进行检查
Base.metadata.tables[User.__table__].create(engine, checkfirst=True)

声明式映射

声明式映射是现代 SQLAlchemy 最典型的映射方式，他通过构造继承类来实现，具体分为两个步骤:

  1. 创建声明基类(DeclarativeBase): 他可以看作是 registry 的类表示形式，使用 registry.generate_base() 是最方便的创建声明基类的方式
  2. 继承声明基类来构造映射类: 他对应了表，其中的属性表示表中的字段

构造声明基类

所谓的声明基类实际上就是用类的方式来表示 registry，他表示一个数据库对象，所有需要注册到数据库的表都应该是这个基类的子类，通过这种继承的关系来自动注册他:

Python
from sqlalchemy.orm import DeclarativeBase

# 使用默认的 DeclarativeBase 超类来构造声明基类
# 这里的 registry 使用默认的 registry 来创建
class Base(DeclarativeBase):
    pass

# 使用现有的 registry 可以
class Base(DeclarativeBase):
    registry = my_own_registry

# 另一种构造 Base 的方式，他与上面的等价的
Base = my_own_registry.generate_base()

# 最终的子类定义
class User(Base):
    pass

# 装饰器模式同样可以，他替代了继承的定义形式
@my_own_registry.mapped
class User:
    pass

映射类

通过继承声明基类来构建映射类:

Python
from datetime import datetime
from typing import Optional

from sqlalchemy import ForeignKey,func,Integer,String
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column,relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user"

    id = mapped_column(Integer, primary_key=True)
    name: Mapped[str]
    fullname: Mapped[Optional[str]]
    nickname: Mapped[Optional[str]] = mapped_column(String(64))
    create_date: Mapped[datetime] = mapped_column(insert_default=func.now())

    addresses: Mapped[List["Address"]] = relationship(back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    user_id = mapped_column(ForeignKey("user.id"))
    email_address: Mapped[str]

    user: Mapped["User"] = relationship(back_populates="addresses")

他实际上等效于 Table:

Python
user_table = Table(
    "user",                     # 转义成 User.__tablename__ 属性表示表名
    Base.metadata,              # 实际上就是 Base，被定义在 registry 中
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("fullname", String()),
    Column("nickname", String(30)),
)

# 还有一种混合声明方式
class User(Base):
    __table__ = Table(
        "user",
        Base.metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String),
        Column("fullname", String),
        Column("nickname", String),
    )

# 甚至一个 SELECT(即子表) 也可以用这种方式来构建映射类
from sqlalchemy import func, select

customer_select = (
    select(customers)
    .where(id>10)
)

class Customer(Base):
    __table__ = customer_select

# 不过这主要在反射表时使用
engine = create_engine("postgresql+psycopg2://user:pass@hostname/my_existing_database")


class Base(DeclarativeBase):
    pass


class MyClass(Base):
    __table__ = Table(
        "mytable",
        Base.metadata,
        autoload_with=engine,
    )

mapped_column: 包装列

映射类的核心在于定义列，这使用 mapped_column() 方法来实现，该方法接受与 Column 相同的属性，他可以看作是 Column 的包装:

Python
mapped_column(
    __name: str,                # 字段名称, 他与类字段名，类字段名在代码中使用，这个就是数据库的字段名如果两者相同
    __type: TypeEngine,         # 数据类型
    nullable=True,              # nullable 约束
    primary_key=False,          # 主键约束
    ...
)

为了能够提供类型提示的功能，还提供了 Mapped[type] 来实现，因此一个完整的现代化列定义:

Python
# Integer 主键
id:Mapped[int] = mapped_column(Integer, primary_key=True)
# String 类型，相当于自动添加了 mapped_column(String, nullable=False)
name: Mapped[str]
# 这个相当于 mapped_column(String)
fullname: Mapped[Optional[str]]
nickname: Mapped[Optional[str]] = mapped_column(String(64))

命令式映射

命令式映射则依赖 Table 对象，之后使用 registry.map_imperatively() 方法进行包装来实现:

Python
from sqlalchemy import Table, Column, Integer, String, ForeignKey
from sqlalchemy.orm import registry

mapper_registry = registry()

user_table = Table(
    "user",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("fullname", String(50)),
    Column("nickname", String(12)),
)

class User:
    pass

mapper_registry.map_imperatively(User, user_table)

他的好处明显就是能够充分利用已有的 Table 对象，但是问题也比较明显就是非常不直观并且无法提供类型检查等功能。