如何在去重作者时维护书籍与作者的多对多关联？

我有三个模型：Author、Book和BookAuthor。BookAuthor模型代表Author与Book之间的多对多关系，即一本书可以有多个作者，一个作者可以写多本书。

需要解析书籍文件并将书籍、作者及其关联关系存储到数据库中，同时移除Author模型中的重复作者（重复定义为first_name、last_name、date_of_birth和date_of_death属性均相同的记录），并更新BookAuthor关联对象中的关系，使用去重后的作者ID。

要求：避免使用Python数据结构处理重复项（数据源包含数百万条记录，内存消耗过大），尽量将工作委托给数据库。

模型与初始代码 #

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

import pandas as pd


class Base(DeclarativeBase):
    pass


class BookAuthor(Base):
    __tablename__ = "book_author"
    book_id: Mapped[int] = mapped_column(ForeignKey("book.id"), primary_key=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("author.id"), primary_key=True)
    book: Mapped["Book"] = relationship(back_populates="authors")
    author: Mapped["Author"] = relationship(back_populates="books")


class Book(Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(2048))
    authors: Mapped[list["BookAuthor"]] = relationship(back_populates="book")


class Author(Base):
    __tablename__ = "author"

    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str] = mapped_column(String(256))
    last_name: Mapped[str] = mapped_column(String(256))

    date_of_birth: Mapped[int] = mapped_column(Integer, nullable=True)
    date_of_death: Mapped[int] = mapped_column(Integer, nullable=True)

    books: Mapped[list["BookAuthor"]] = relationship(back_populates="author")


engine = create_engine("sqlite:///books.db")

Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

books = [
    {
        "title": "Good Omens: The Nice and Accurate Prophecies of Agnes Nutter, Witch",
        "authors": [
            {
                "first_name": "Terry",
                "last_name": "Pratchett",
                "date_of_birth": 1948,
                "date_of_death": 2015,
            },
            {
                "first_name": "Neil",
                "last_name": "Gaiman",
                "date_of_birth": 1960,
                "date_of_death": None,
            },
        ],
    },
    {
        "title": "American Gods",
        "authors": [
            {
                "first_name": "Neil",
                "last_name": "Gaiman",
                "date_of_birth": 1960,
                "date_of_death": None,
            },
        ],
    },
    {
        "title": "The Talisman",
        "authors": [
            {
                "first_name": "Stephen",
                "last_name": "King",
                "date_of_birth": 1947,
                "date_of_death": None,
            },
            {
                "first_name": "Peter",
                "last_name": "Straub",
                "date_of_birth": 1943,
                "date_of_death": 2022,
            },
        ],
    },
    {
        "title": "The Shining",
        "authors": [
            {
                "first_name": "Stephen",
                "last_name": "King",
                "date_of_birth": 1947,
                "date_of_death": None,
            },
        ],
    },
    {
        "title": "It",
        "authors": [
            {
                "first_name": "Stephen",
                "last_name": "King",
                "date_of_birth": 1947,
                "date_of_death": None,
            },
        ],
    },
]

# 初始写入代码（会产生重复作者）
with Session(engine) as session:
    for book in books:
        book_db = Book(title=book["title"])
        for author in book["authors"]:
            author_db = Author(
                first_name=author["first_name"],
                last_name=author["last_name"],
                date_of_birth=author["date_of_birth"],
                date_of_death=author["date_of_death"],
            )
            book_author = BookAuthor(book=book_db, author=author_db)
            session.add(book_author)

    session.commit()

    # 查询输出
    book_authors = str(
        session.query(BookAuthor)
        .join(Book)
        .join(Author)
        .with_entities(
            BookAuthor.book_id.label("book_id"),
            BookAuthor.author_id.label("author_id"),
            Author.first_name,
            Author.last_name,
            Author.date_of_birth,
            Author.date_of_death,
            Book.title,
        )
    )

    print(pd.read_sql_query(sql=book_authors, con=engine).to_string(index=False))

当前输出（存在重复作者） #

+---------+-----------+-------------------+------------------+----------------------+----------------------+---------------------------------------------------------------------+
| book_id | author_id | author_first_name | author_last_name | author_date_of_birth | author_date_of_death |                             book_title                              |
+---------+-----------+-------------------+------------------+----------------------+----------------------+---------------------------------------------------------------------+
|       1 |         1 | Terry             | Pratchett        |                 1948 |               2015.0 | Good Omens: The Nice and Accurate Prophecies of Agnes Nutter, Witch |
|       1 |         2 | Neil              | Gaiman           |                 1960 |                      | Good Omens: The Nice and Accurate Prophecies of Agnes Nutter, Witch |
|       2 |         3 | Neil              | Gaiman           |                 1960 |                      | American Gods                                                       |
|       3 |         4 | Stephen           | King             |                 1947 |                      | The Talisman                                                        |
|       3 |         5 | Peter             | Straub           |                 1943 |               2022.0 | The Talisman                                                        |
|       4 |         6 | Stephen           | King             |                 1947 |                      | The Shining                                                         |
|       5 |         7 | Stephen           | King             |                 1947 |                      | It                                                                  |
+---------+-----------+-------------------+------------------+----------------------+----------------------+---------------------------------------------------------------------+

期望输出（去重后） #

+---------+-----------+-------------------+------------------+----------------------+----------------------+---------------------------------------------------------------------+
| book_id | author_id | author_first_name | author_last_name | author_date_of_birth | author_date_of_death |                             book_title                              |
+---------+-----------+-------------------+------------------+----------------------+----------------------+---------------------------------------------------------------------+
|       1 |         1 | Terry             | Pratchett        |                 1948 |               2015.0 | Good Omens: The Nice and Accurate Prophecies of Agnes Nutter, Witch |
|       1 |         2 | Neil              | Gaiman           |                 1960 |                      | Good Omens: The Nice and Accurate Prophecies of Agnes Nutter, Witch |
|       2 |         2 | Neil              | Gaiman           |                 1960 |                      | American Gods                                                       |
|       3 |         3 | Stephen           | King             |                 1947 |                      | The Talisman                                                        |
|       3 |         4 | Peter             | Straub           |                 1943 |               2022.0 | The Talisman                                                        |
|       4 |         3 | Stephen           | King             |                 1947 |                      | The Shining                                                         |
|       5 |         3 | Stephen           | King             |                 1947 |                      | It                                                                  |
+---------+-----------+-------------------+------------------+----------------------+----------------------+---------------------------------------------------------------------+

针对百万级数据，推荐两种数据库层面的处理方案，避免Python内存开销：

方案一：写入阶段避免重复作者（推荐） #

从源头阻止重复作者写入，结合数据库唯一约束和批量插入，效率最高。

步骤1：给Author模型添加唯一约束 #

确保数据库层面不会插入重复作者：

class Author(Base):
    __tablename__ = "author"

    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str] = mapped_column(String(256))
    last_name: Mapped[str] = mapped_column(String(256))
    date_of_birth: Mapped[int] = mapped_column(Integer, nullable=True)
    date_of_death: Mapped[int] = mapped_column(Integer, nullable=True)

    books: Mapped[list["BookAuthor"]] = relationship(back_populates="author")

    # 添加唯一约束，确保四个字段组合唯一
    __table_args__ = (
        UniqueConstraint('first_name', 'last_name', 'date_of_birth', 'date_of_death', name='unique_author'),
    )

步骤2：批量插入作者+关联书籍 #

使用SQLAlchemy的批量插入语法，忽略重复作者，再关联书籍：

from sqlalchemy.dialects.sqlite import insert  # 不同数据库语法略有差异，比如PostgreSQL用on_conflict_do_nothing

with Session(engine) as session:
    # 提取所有作者数据（无需去重，数据库会处理）
    all_authors = []
    for book in books:
        all_authors.extend(book["authors"])
    
    # 批量插入作者，已存在则忽略
    insert_stmt = insert(Author).values(all_authors)
    insert_stmt = insert_stmt.on_conflict_do_nothing(constraint='unique_author')
    session.execute(insert_stmt)
    
    # 插入书籍并关联已有作者
    for book in books:
        book_db = Book(title=book["title"])
        session.add(book_db)
        session.flush()  # 刷入数据库获取book_id，便于关联
        
        for author in book["authors"]:
            # 查询已存在的作者
            author_db = session.query(Author).filter(
                Author.first_name == author["first_name"],
                Author.last_name == author["last_name"],
                Author.date_of_birth == author["date_of_birth"],
                Author.date_of_death == author["date_of_death"]
            ).first()
            # 创建关联记录
            book_author = BookAuthor(book_id=book_db.id, author_id=author_db.id)
            session.add(book_author)
    
    session.commit()

方案二：事后数据库层面去重清理 #

如果已经写入了重复数据，直接用SQL语句批量处理，全程由数据库执行：

执行SQL去重（以SQLite为例） #

with Session(engine) as session:
    # 1. 创建临时表，存储每个重复作者组的主ID（保留最小ID）
    session.execute("""
        CREATE TEMP TABLE author_main_ids AS
        SELECT MIN(id) as main_id, first_name, last_name, date_of_birth, date_of_death
        FROM author
        GROUP BY first_name, last_name, date_of_birth, date_of_death;
    """)
    
    # 2. 更新关联表，将重复作者ID替换为主ID
    session.execute("""
        UPDATE book_author
        SET author_id = (
            SELECT main_id
            FROM author_main_ids
            WHERE author.first_name = author_main_ids.first_name
              AND author.last_name = author_main_ids.last_name
              AND author.date_of_birth = author_main_ids.date_of_birth
              AND (author.date_of_death = author_main_ids.date_of_death OR (author.date_of_death IS NULL AND author_main_ids.date_of_death IS NULL))
        )
        WHERE author_id IN (
            SELECT id
            FROM author
            WHERE id NOT IN (SELECT main_id FROM author_main_ids)
        );
    """)
    
    # 3. 删除重复的作者记录
    session.execute("""
        DELETE FROM author
        WHERE id NOT IN (SELECT main_id FROM author_main_ids);
    """)
    
    # 清理临时表
    session.execute("DROP TABLE author_main_ids;")
    session.commit()

内容的提问来源于stack exchange，提问作者Thegerdfather