• SqlAlchemy使用教程(一) 原理與環(huán)境搭建
• SqlAlchemy使用教程(二) 入門(mén)示例及編程步驟
• SqlAlchemy使用教程(三) CoreAPI訪問(wèn)與操作數(shù)據(jù)庫(kù)詳解
• SqlAlchemy使用教程(四) MetaData 與 SQL Express Language 的使用
• SqlAlchemy使用教程(五) ORM API 編程入門(mén)

本章內(nèi)容，稍微有些復(fù)雜，建議騰出2小時(shí)空閑時(shí)間，沖杯咖啡或泡杯茶 ?? , 慢慢看，在電腦上跑下代碼，可以加深理解.

六、表間關(guān)系的定義與CRUD操作

表間關(guān)系主要包括：一對(duì)多，一對(duì)一，多對(duì)多。其中一對(duì)多關(guān)系中也隱含了多對(duì)一關(guān)系。
表間關(guān)系是數(shù)據(jù)庫(kù)操作中的重要技術(shù)點(diǎn)，非常有必要理解與掌握。

1、 一對(duì)多表間關(guān)系的定義

一對(duì)多表間關(guān)系實(shí)現(xiàn)語(yǔ)法

以一對(duì)多關(guān)系為例，

• 子表側(cè)，與父表是一對(duì)多關(guān)系，
• 父表側(cè)，可以反向查詢(xún)子表數(shù)據(jù)，與子表是多對(duì)一關(guān)系。

class Parent(Base):
    __tablename__ = "parent_table"
    id: Mapped[int] = mapped_column(primary_key=True)
    children: Mapped[List["Child"]] = relationship(back_populates="parent")


class Child(Base):
    __tablename__ = "child_table"
    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int] = mapped_column(ForeignKey("parent_table.id"))
    parent: Mapped["Parent"] = relationship(back_populates="children")

說(shuō)明：
1）在子表中添加外鍵字段，以及relationship()引用，
2）在父表中添加relationship()引用，用于反向查詢(xún)。

示例代碼

父表：company, 子表 person, 表結(jié)構(gòu)類(lèi)定義如下。

from sqlalchemy.orm import DeclarativeBase, Session
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy import ForeignKey
from sqlalchemy import String, Integer
from typing import List

class Base(DeclarativeBase):
    pass
class Company(Base):
    __tablename__ = "company"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    company_name: Mapped[str] = mapped_column(String(30), index=True)
    persons: Mapped[List['Person']] = relationship(
        back_populates="company", cascade="all, delete-orphan")

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

class Person(Base):
    __tablename__ = "person"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    age: Mapped[int] = mapped_column(Integer)
    company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
    company: Mapped['Company'] = relationship(back_populates="persons")

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

說(shuō)明：
1）從子表視角看，1個(gè)人只屬于1個(gè)Company; 但1個(gè)Company 對(duì)應(yīng)多個(gè)人，因此在父表則，relationship() 左側(cè)的類(lèi)型注解為 List[‘Person’], 也可以用Set[‘Person’]
2）父表中添加刪除依賴(lài)，cascade=“all, delete-orphan”，即子表中不存在對(duì)父表記錄的引用時(shí)，才能刪除，以保證數(shù)據(jù)的完整性。
3）當(dāng)前版本可能存在bug, 官方文檔中的示例中有的字段使用簡(jiǎn)化寫(xiě)法（右側(cè)未給出mapped_column())，sqlite3運(yùn)行是沒(méi)有問(wèn)題的，但mysql, postgresql創(chuàng)建表時(shí)會(huì)丟棄簡(jiǎn)化寫(xiě)法的字段，導(dǎo)致后續(xù)insert等操作失敗。 因此請(qǐng)嚴(yán)格請(qǐng)勿采有簡(jiǎn)化寫(xiě)法。

多對(duì)一關(guān)系的實(shí)現(xiàn)語(yǔ)法

當(dāng)不需要反向查詢(xún)時(shí)，則父表與子表形成Many to One 多對(duì)一關(guān)系， 在父表則添加子表的外鍵與relationship()引用，子表無(wú)須做額外配置

class Parent(Base):
    __tablename__ = "parent_table"
    id: Mapped[int] = mapped_column(primary_key=True)
    child_id: Mapped[int] = mapped_column(ForeignKey("child_table.id"))
    child: Mapped["Child"] = relationship()


class Child(Base):
    __tablename__ = "child_table"
    id: Mapped[int] = mapped_column(primary_key=True)

如果允許 child_id空值，則將改字段的類(lèi)型注解修改為

from typing import Optional
......
child_id: Mapped[Optional[int]] = mapped_column(ForeignKey("child_table.id"))

對(duì)于3.10+版本，類(lèi)型注解支持 | 操作符

child_id: Mapped[int | None] = mapped_column(ForeignKey("child_table.id"))
child: Mapped[Child | None] = relationship(back_populates="parents")

2、 插入數(shù)據(jù)與多表聯(lián)合查詢(xún)

1）在數(shù)據(jù)庫(kù)創(chuàng)建表

# 創(chuàng)建數(shù)據(jù)庫(kù)連接引擎對(duì)象
engine = create_engine(
    "mysql+mysqlconnector://root:Admin&123@localhost:3306/testdb")
# 將DDL語(yǔ)句映射到數(shù)據(jù)庫(kù)表，如果數(shù)據(jù)庫(kù)表不存在，則創(chuàng)建該表
Base.metadata.create_all(engine)
# 打印創(chuàng)建創(chuàng)建的表
Print(Base.metadata.tables) 

2）插入數(shù)據(jù)

基本步驟包括：

(1)為測(cè)試方便，先寫(xiě)1個(gè)get_or_create()函數(shù)，如果插入對(duì)象在數(shù)據(jù)庫(kù)中已存在則不插入，以方便連續(xù)測(cè)試。
(2)創(chuàng)建session對(duì)象
(3)先創(chuàng)建父表對(duì)象并插入
(4)創(chuàng)建子表對(duì)象并插入
(5)用多表查詢(xún)方法檢查結(jié)果

Step-1: 自定義create_or_create()函數(shù)

官方提供的upsert方法不通用。下面函數(shù)是通用的，

def get_or_create(session, model, defaults=None, **kwargs):
    """如果不存在則創(chuàng)建，如果存在則返回
    輸入?yún)?shù)：
        session: sqlalchemy session
        model: 自定義的ORM類(lèi)
        defaults: 有默認(rèn)值的字段
        kwargs: 其他字段(必須包含主要字段)
    返回值：
        instance: 返回的實(shí)例
    """
    instance = session.query(model).filter_by(**kwargs).first()
    if instance:
        print("instance already exists", instance)
        return instance
    else:
        params = dict((k, v) for k, v in kwargs.items()
                      if not isinstance(v, ClauseElement))
        if defaults:
            params.update(defaults)
        instance = model(**params)
        session.add(instance)
        session.commit()
        print("instance inserted", instance)
        return instance

Step-2: 向兩個(gè)關(guān)聯(lián)表插入數(shù)據(jù)

用with 語(yǔ)句創(chuàng)建session對(duì)象，插入操作順序，先父表再子表

with Session(engine) as session:
    # 插入數(shù)據(jù)
    get_or_create(session, Company, company_name="蜀漢")
    get_or_create(session, Company, company_name="曹魏")
    get_or_create(session, Company, company_name="東吳")
    
    stmt = select(Company)
    results = session.scalars(stmt)
    print(results.all())

    # insert data in person table
    company_shu = session.scalars(select(Company).where(
        Company.company_name == "蜀漢")).first()
    get_or_create(session, Person, name="劉備",
                  age=42, company=company_shu)
    get_or_create(session, Person, name="關(guān)羽",
                  age=40, company=company_shu)
    get_or_create(session, Person, name="張飛", age=38, company=company_shu)
    company_wei = session.scalars(select(Company).where(
        Company.company_name == "曹魏")).first()
    get_or_create(session, Person, name="張遼", age=40, company=company_wei)
    get_or_create(session, Person, name="曹操", age=38, company=company_wei)
    company_wu = session.scalars(select(Company).where(
        Company.company_name == "東吳")).first()
    get_or_create(session, Person, name="周瑜", age=30, company=company_wu)

3) 多表聯(lián)合查詢(xún)

   # select with Join 多表查詢(xún)
    stmt = select(Person).join(Person.company).where(
        Company.company_name == "蜀漢").order_by(Person.age)
    results = session.scalars(stmt)
    # 遍歷結(jié)果
    for r in results:
        print(r.name, r.age, r.company.company_name)

Output:

instance inserted Company(id=1, company_name=蜀漢)
instance inserted Company(id=2, company_name=曹魏)
instance inserted Company(id=3, company_name=東吳)
[Company(id=1, company_name=蜀漢), Company(id=2, company_name=曹魏), Company(id=3, company_name=東吳)]
instance inserted Person(id=1, name=劉備)
instance inserted Person(id=2, name=關(guān)羽)
instance inserted Person(id=3, name=張飛)
instance inserted Person(id=4, name=張遼)
instance inserted Person(id=5, name=曹操)
instance inserted Person(id=6, name=周瑜)
張飛 38 蜀漢
關(guān)羽 40 蜀漢
劉備 42 蜀漢

刪除數(shù)據(jù)
當(dāng)刪除父表記錄時(shí)，子表中應(yīng)無(wú)對(duì)此數(shù)據(jù)的引用，否則無(wú)法刪除。

3、 一對(duì)一關(guān)系

從外鍵角度看，一對(duì)一關(guān)系也是一對(duì)多關(guān)系。實(shí)現(xiàn)時(shí)，

• 在父表中收集子表數(shù)據(jù)時(shí)，類(lèi)型注解不使用集合類(lèi)型即可。
• 子表中relationship()方法中添加single_parent=True.

class Parent(Base):
    __tablename__ = "parent_table"
    id: Mapped[int] = mapped_column(primary_key=True)
child: Mapped["Child"] = relationship(back_populates="parent")   
    # 一對(duì)多時(shí)，使用child: Mapped[List["Child"]] 


class Child(Base):
    __tablename__ = "child_table"
    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int] = mapped_column(ForeignKey("parent_table.id"))
    parent: Mapped["Parent"] = relationship(back_populates="child"，single_parent=True)

4、 多對(duì)多關(guān)系

多對(duì)多關(guān)系特點(diǎn)：
1）父表與子表，子表與父表之間均為多對(duì)多關(guān)系。
2）通常使用1張中間表， 與父表、子表均實(shí)現(xiàn)1對(duì)多關(guān)系。
（當(dāng)然有的ORM模型將中間表的創(chuàng)建隱藏起來(lái)，但在數(shù)據(jù)庫(kù)中還是可以看到）

多對(duì)多關(guān)系定義示例

from __future__ import annotations

from sqlalchemy import Column
from sqlalchemy import Table
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


# note for a Core table, we use the sqlalchemy.Column construct,
# not sqlalchemy.orm.mapped_column
association_table = Table(
    "association_table",
    Base.metadata,
    Column("left_id", ForeignKey("left_table.id")),
    Column("right_id", ForeignKey("right_table.id")),
)


class Parent(Base):
    __tablename__ = "left_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    children: Mapped[List[Child]] = relationship(
        secondary=association_table, back_populates="parents"
    )


class Child(Base):
    __tablename__ = "right_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    parents: Mapped[List[Parent]] = relationship(
        secondary=association_table, back_populates="children"
    )

多對(duì)多關(guān)系的查詢(xún)、插入操作與一對(duì)多查詢(xún)相似。 需要注意的是刪除操作。

從多對(duì)多關(guān)系中刪除數(shù)據(jù)

用SQL來(lái)實(shí)現(xiàn)時(shí)，需要先從父表與子表刪除數(shù)據(jù)，再?gòu)闹虚g表刪除。ORM API 可以自動(dòng)完成這個(gè)過(guò)程。 如要?jiǎng)h除子表的某條記錄。

myparent.children.remove(somechild)

注：通過(guò)session.delete(somechild)時(shí)，MySql可能會(huì)報(bào)錯(cuò)，我遇到的原因有多種，不建議使用。
同樣，如果要?jiǎng)h除父表中的1條記錄：文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-817512.html

mychild.parent.remove(someparent) 

到了這里，關(guān)于SqlAlchemy使用教程(六) -- ORM 表間關(guān)系的定義與CRUD操作的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！