本文编写时 SQLAlchemy 版本为 2.0.35

安装

SQLAlchemy

pip install SQLAlchemy

MySQL 客户端（驱动）

参考：MySQL and MariaDB — SQLAlchemy 2.0 Documentation (opens new window)

连接

from sqlalchemy import create_engine
# sqlite
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)

# mysql
engine = create_engine("mysql+mysqldb://scott:tiger@localhost/test")

直接执行

from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("select 'hello world'"))
    print(result.all())


# 多行结果
with engine.connect() as conn:
    result = conn.execute(text("SELECT x, y FROM some_table"))
    for row in result:
        print(f"x: {row.x}  y: {row.y}")

# 插入，需要手动commit
with engine.connect() as conn:
    conn.execute(text("CREATE TABLE some_table (x int, y int)"))
    conn.execute(
        text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
        [{"x": 1, "y": 1}, {"x": 2, "y": 4}],
    )
    conn.commit()


# 查询参数
with engine.connect() as conn:
    result = conn.execute(text("SELECT x, y FROM some_table WHERE y > :y"), {"y": 2})
    for row in result:
        print(f"x: {row.x}  y: {row.y}")

# 多参数
with engine.connect() as conn:
    conn.execute(
        text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
        [{"x": 11, "y": 12}, {"x": 13, "y": 14}],
    )
    conn.commit()

ORM 操作

上述的例子均为直接操作 SQL，ORM 的 O 便是 Object，因此引入此库更多是为了操作对象。

ORM Session

使用 Session(engine) 代替 engine.connect()

笔记

暂不清楚是作用是什么，可能是因为 connect 操作是每次操作都要创建、释放连接，而 Session 是使用池化技术？待考证……

from sqlalchemy.orm import Session

stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y")
with Session(engine) as session:
    result = session.execute(stmt, {"y": 6})
    for row in result:
        print(f"x: {row.x}  y: {row.y}")

定义模型

创建一个包 model ，在包下根据模块创建对应存放模型的 py 文件，如 order.py 、 pay.py 等，如果没有细分模块可以直接叫 model.py 。

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    # 定义一个基类，继承自DeclarativeBase
    pass

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

参考

• SQLAlchemy Documentation — SQLAlchemy 2.0 Documentation (opens new window)