使用数据库元数据
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
文章评论