Sqlalchemy WriteOnly(select，joined)详解，不止只写！

站长

2023年08月28日 20:40 · 阅读数 15

自从 Sqlalchemy2.0 版本以来，更新了很多东西，以后可以好好讲讲。

WriteOnly

WriteOnly 是 relationship 加载模式中的一员，通常用于一对多或者多对多关系。是旧版 dynamic 加载器的继任者。与其他加载器不同的地方在于，它原生支持异步，也就是说，不需要通过继承 AsyncAttr，或者使用 awaitable_attr 来访问。旧版的 dynamic 则不支持异步。

虽然他的名字叫 只写，但是，实际上并不是只起到一个只写属性的作用，只是属性还没被加载之前，只可以写 relationship 相关数据，如果需要用到其相关的数据，需要单独发送 ORM 请求。

官方文档：

Working with Large Collections — SQLAlchemy 2.0 Documentation --- 使用大型集合 — SQLAlchemy 2.0 文档

本文将在 asyncio 的环境下，对文档中的内容做提取抽象，以及补充一些常用的用法。

我们将会通过一个例子，来展示 lazy = write_only , select, joined 三者之间的区别，以及讲解关于 write_only loader 的用法。

定义关系

from typing import ClassVar

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncAttrs
from sqlalchemy.orm import DeclarativeBase, Mapped, WriteOnlyMapped, mapped_column, relationship
from sqlalchemy.sql.schema import Table, Column, ForeignKey
from sqlalchemy.sql.sqltypes import Text, Integer

# 基于内存的 engine
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=True)
make_session = async_sessionmaker(bind=engine, expire_on_commit=False, autoflush=True)


class Base(DeclarativeBase, AsyncAttrs):
    ...


# 创建一个两者的中继表
student_lesson = Table(
    "student->lesson",
    Base.metadata,
    Column("student_id", Integer, ForeignKey("student.id")),
    Column("lesson_id", Integer, ForeignKey("lesson.id")),
)


class Student(Base):
    __tablename__ = "student"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(Text(64))

    # 为 relationship 指定外键，让 ORM 知道如何关联
    class_id: Mapped[int] = mapped_column(Integer, ForeignKey("class.id"))

    # 单个的 relationship，使用 typing.ClassVar 来指定类型
    # 如果访问 Student 时，基本都会使用到 Class 的数据，那么使用 joined 是最好的选择
    class_: ClassVar["Class"] = relationship(
        "Class",
        lazy="joined",
        uselist=False
    )

    # 一个学生可以有多个课程, 一个课程可以有多个学生
    # 所以此处使用 secondary 来指定中继表
    lessons: WriteOnlyMapped["Lesson"] = relationship(
        "Lesson",
        secondary=student_lesson,
        lazy="write_only",
    )


class Class(Base):
    __tablename__ = "class"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(Text(64))

    students: ClassVar[Student] = relationship(
        "Student",
        lazy="select",
        uselist=True,
    )


class Lesson:
    __tablename__ = "lesson"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(Text(64))

    students: WriteOnlyMapped[Student] = relationship(
        "Student",
        secondary=student_lesson,
        lazy="write_only",
    )
    
if __name__ == '__main__':
    from IPython import start_ipython
    # 进入 iPython 环境调试
    start_ipython(argv=[], user_ns=locals())

创建数据库(使用异步)

async def create_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


async def drop_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

准备数据

开始之前我们先准备一些数据，以便待会使用它。

async def test():
    async with make_session() as session:
        # 创建两个班级
        class1 = Class(name="大专")
        class2 = Class(name="本科")
        session.add_all([class1, class2])

        # 创建三个学生
        z3 = Student(name="张三", class_id=1)
        l3 = Student(name="李四", class_id=1)
        w5 = Student(name="王五", class_id=2)
        session.add_all([z3, l3, w5])

        # 创建两个课程
        math = Lesson(name="数学")
        english = Lesson(name="英语")
        session.add_all([math, english])

        await session.commit()