SQAlchemy 第二篇

使用数据库元数据

SQLAlchemy 中数据库元数据最常见的基础对象称为 MetaData、Table和Column。下面的部分将说明如何在面向 Core 的风格和面向 ORM 的风格中使用这些对象。

使用表对象设置元数据

当我们使用关系数据库时，我们查询的数据库中的基本数据保存结构称为表。在 SQLAlchemy 中，数据库“表”最终由一个类似名称的 Python 对象表示Table。

要开始使用 SQLAlchemy 表达式语言，我们需要 Table构建代表我们感兴趣的所有数据库表的对象。以编程方式构造Table，可以直接使用 Table构造函数，也可以间接使用 ORM 映射类（稍后将在使用 ORM 声明式表单定义表元数据中进行描述）。还有一个选项可以从现有数据库加载部分或全部表信息，称为反射。

无论使用哪种方法，我们总是从一个集合开始，该集合将是我们放置表（称为对象）的地方MetaData 。该对象本质上是围绕 Python 字典的外观，该字典存储一系列Table以其字符串名称为键的对象。虽然 ORM 提供了一些关于从哪里获取此集合的选项，但我们始终可以选择直接创建一个集合，如下所示：

from sqlalchemy import MetaData

metadata_obj = MetaData()

一旦我们有了一个MetaData对象，我们就可以声明一些 Table对象。本教程将从经典的 SQLAlchemy 教程模型开始，该模型有一个名为 的表user_account，用于存储网站的用户等信息，以及一个相关表address，用于存储与表中的行关联的电子邮件地址user_account 。当根本不使用 ORM 声明性模型时，我们 Table直接构造每个对象，通常将每个对象分配给一个变量，这将是我们在应用程序代码中引用表的方式：

from sqlalchemy import Table, Column, Integer, String, MetaData

metadata_obj = MetaData()
user_table = Table(
    "user_account",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("name", String(255)),
    Column("fullname", String(255)),
)
user_table.c.name

Column('name', String(length=255), table=<user_account>)

user_table.c.keys()

['id', 'name', 'fullname']

user_table.primary_key

PrimaryKeyConstraint(Column('id', Integer(), table=<user_account>, primary_key=True, nullable=False))

声明外间约束

from sqlalchemy import ForeignKey

address_table = Table(
    "address",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("user_id", ForeignKey("user_account.id"), nullable=False),
    Column("email_address", String(255), nullable=False),
)
address_table.c.id

Column('id', Integer(), table=<address>, primary_key=True, nullable=False)

执行DDL操作

from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:root@localhost/test?charset=utf8mb4", echo=True, echo_pool=True,
                       pool_size=20)
metadata_obj.create_all(engine)

2023-12-14 13:30:57,021 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2023-12-14 13:30:57,021 INFO sqlalchemy.engine.Engine [raw sql] {}
2023-12-14 13:30:57,024 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2023-12-14 13:30:57,024 INFO sqlalchemy.engine.Engine [raw sql] {}
2023-12-14 13:30:57,025 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2023-12-14 13:30:57,026 INFO sqlalchemy.engine.Engine [raw sql] {}
2023-12-14 13:30:57,028 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-12-14 13:30:57,029 INFO sqlalchemy.engine.Engine DESCRIBE `test`.`user_account`
2023-12-14 13:30:57,029 INFO sqlalchemy.engine.Engine [raw sql] {}
2023-12-14 13:30:57,031 INFO sqlalchemy.engine.Engine DESCRIBE `test`.`address`
2023-12-14 13:30:57,031 INFO sqlalchemy.engine.Engine [raw sql] {}
2023-12-14 13:30:57,033 INFO sqlalchemy.engine.Engine 
CREATE TABLE user_account (
	id INTEGER NOT NULL AUTO_INCREMENT, 
	name VARCHAR(255), 
	fullname VARCHAR(255), 
	PRIMARY KEY (id)
)


2023-12-14 13:30:57,034 INFO sqlalchemy.engine.Engine [no key 0.00097s] {}
2023-12-14 13:30:57,061 INFO sqlalchemy.engine.Engine 
CREATE TABLE address (
	id INTEGER NOT NULL AUTO_INCREMENT, 
	user_id INTEGER NOT NULL, 
	email_address VARCHAR(255) NOT NULL, 
	PRIMARY KEY (id), 
	FOREIGN KEY(user_id) REFERENCES user_account (id)
)


2023-12-14 13:30:57,062 INFO sqlalchemy.engine.Engine [no key 0.00055s] {}
2023-12-14 13:30:57,086 INFO sqlalchemy.engine.Engine COMMIT

使用orm的声明式方式定义表元数据

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


Base.metadata

MetaData()

Base.registry

<sqlalchemy.orm.decl_api.registry at 0x1191d78d0>

声明映射类

from typing import List, Optional
from sqlalchemy.orm import Mapped, mapped_column, relationship


class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(20))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["address"]] = relationship("user")

    def __repr__(self) -> str:
        return f"User(id={self.id},name={self.name})"


class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id = mapped_column(ForeignKey("user_account.id"))
    user: Mapped[User] = relationship(back_populates="addresses")

    def __repr__(self) -> str:
        return f"Address(id={self.id},email_address={self.email_address})"


Base.metadata.create_all(engine)

2023-12-14 13:46:18,994 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-12-14 13:46:18,995 INFO sqlalchemy.engine.Engine DESCRIBE `test`.`user_account`
2023-12-14 13:46:18,996 INFO sqlalchemy.engine.Engine [raw sql] {}
2023-12-14 13:46:18,998 INFO sqlalchemy.engine.Engine DESCRIBE `test`.`address`
2023-12-14 13:46:18,998 INFO sqlalchemy.engine.Engine [raw sql] {}
2023-12-14 13:46:19,000 INFO sqlalchemy.engine.Engine COMMIT