SQLAlchemy一对多关系持久化异常：关联对象仅最后一条生效

环境 #

• sqlalchemy 2.0.23
• postgres 16.1
• python 3.11.6

问题现象 #

在Publication与EducationalSystem模型间建立多对一关系后，循环为多个Publication实例分配同一个EducationalSystem实例时，仅最后一个Publication的外键educational_system_fk被正确赋值，其余实例的该字段均为null。

模型代码 #

class Publication(Base):
    # __tablename__ = "publication"
    __table_args__ = (
        UniqueConstraint("section_id", "title", name="section_title_uc"),
        { "schema" : "publications", }
    )
    
    __mapper_args__ = {
        "polymorphic_identity": "publication",
        "polymorphic_on": "publication_type"
    }

    publication_type: Mapped[str] = mapped_column(String(), nullable=False)
    id = mapped_column(Integer, primary_key=True)
    title = mapped_column(String(), nullable=False)
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    updated_at: Mapped[datetime] = mapped_column(onupdate=func.now(), nullable=True)
    section_id: Mapped[int] = mapped_column(ForeignKey(Section.__table__.c.name), nullable=True)
    section = relationship("Section", back_populates="publications")
#### EducationalSystem ForeignKey ########
    educational_system_fk: Mapped[str] = mapped_column(ForeignKey("publications.educational_system.name"), nullable=True)

##### ED SYSTEM relationship object ####
    ed_system: Mapped[EducationalSystem] = relationship(back_populates="publications")
######
    unique_title_section = UniqueConstraint("section_id", "title", "draft", name="utsc") #There can only be a published article in the same section
    schedule: Mapped[datetime] = mapped_column(default=None, nullable=True)
    content: Mapped[str] = mapped_column(nullable=True)
    draft: Mapped[bool] = mapped_column(default=True)
    summary: Mapped[str] = mapped_column(nullable=True)
    quotes: Mapped[list[Quote]] = relationship(back_populates="publication")
    important: Mapped[bool] = mapped_column(nullable=True, default=False)
    views: Mapped[str] = mapped_column(default=0)
    statement: Mapped[str] = mapped_column(nullable=True)
    authors: Mapped[list[Author]] = relationship(secondary=publication_authors_table, back_populates="publications")
    images: Mapped[Image] = relationship(secondary=publication_image_table)
    editor_fk: Mapped[str] = mapped_column(ForeignKey("users.editor.username"), nullable=True)
    editor: Mapped[Editor] = relationship(back_populates="publications")
    

    #synonyms
    educational_system = synonym("ed_system")

    def __repr__(self):
        return f"{self.__class__.__name__}({self.title!r})"


class EducationalSystem(Base):
    __table_args__ = (
        {"schema": "publications"},
    )
    name: Mapped[str] = mapped_column(primary_key=True)
    publications: Mapped[Publication] = relationship(back_populates="ed_system")

初始化函数代码 #

@asynccontextmanager
async def initialize(app: FastAPI):
    with SessionLocal(autoflush=True) as session:
        Base.metadata.create_all(bind=engine)
        stmts = [
            "INSERT INTO users.role VALUES('admin') ON CONFLICT DO NOTHING;"
            "INSERT INTO users.role VALUES('editor') ON CONFLICT DO NOTHING;"
            "INSERT INTO users.role VALUES('normal') ON CONFLICT DO NOTHING;"
            "INSERT INTO publications.section VALUES('news') ON CONFLICT DO NOTHING;"
            "INSERT INTO publications.section VALUES('inquiries') ON CONFLICT DO NOTHING;"
            "INSERT INTO publications.section VALUES('resources') ON CONFLICT DO NOTHING;"
            "INSERT INTO publications.section VALUES('experiences') ON CONFLICT DO NOTHING;"
        ]
        
        for stmt in stmts:
            session.execute(text(stmt))

        users = None
        pubs = None
        authors = None
        ed_systems = None
        quotes = None
        statements = None
        
        with open("./users.json") as f: users = json.load(f)
        with open("./publication.json") as f: pubs = json.load(f)
        with open("./authors.json") as f: authors = json.load(f)
        with open("./educational_systems.json") as f: ed_systems = json.load(f)
        with open("./quotes.json") as f: quotes = json.load(f)
        with open("./statements.json") as f: statements = json.load(f)
        
        editor_objs = []

        for user in users:
            class_ = eval(user["role_fk"].capitalize())
            user_obj = class_(**user)
            if isinstance(user_obj, Editor): editor_objs.append(user_obj)
            session.add(user_obj)
        try: session.commit() #commit users
        except sqlalchemy.exc.IntegrityError as err: print(err)

        author_list = []
        for author in authors:
            author["social_media_handles"] = [SocialMediaHandle(handle=handle) for handle in author["social_media_handles"]] 
            author_obj = Author(**author)
            session.add(author_obj)
            author_list.append(author_obj)
        
        ed_system_objs = []
        
        ###### WHERE EducationalSystems created ######
        for ed_system in ed_systems:
            ed_system_obj = EducationalSystem(name=ed_system)
            ed_system_objs.append(ed_system_obj)
            session.add(ed_system_obj)
            session.commit()

        
        for pub in pubs:
            class_ = eval(
                coerce_genre(pub["publication_type"], to="camel_case").pop()
            )
            pub["authors"] = choices(author_list, k=randint(1,3))
            pub_obj = class_(**pub)
            
            pub_obj.quotes.extend([Quote(**quote) for quote in choices(quotes, k=randint(1,3))])
            pub_obj.statement = statements[randint(0, len(statements) -1)]
            pub_obj.summary = pub_obj.statement
            ## ADDING a same educationalsystem to every publication ###
            pub_obj.ed_system = ed_system_objs[0]
            pub_obj.editor = editor_objs[randint(0, len(editor_objs) - 1)]
            session.add(pub_obj)
            session.commit() # commit
            
        
        
        
        try: session.commit()
        except sqlalchemy.exc.IntegrityError as err: print("Initial registries already exist, SKIPPING!")
        except Exception as err: print(f"{err}")
    yield
    with SessionLocal.begin() as session:
        print("DROPING DATABASE!")
        Base.metadata.drop_all(bind=engine)

解决方案 #

1. 修正关系定义 #

EducationalSystem中的publications字段类型错误，多对一关系中，一方的集合应定义为list[Publication]，否则SQLAlchemy无法正确维护关联：

# 修改EducationalSystem类中的publications字段
publications: Mapped[list[Publication]] = relationship(back_populates="ed_system")

2. 避免频繁提交Session #

循环内每次创建Publication后调用session.commit()会破坏Unit of Work的状态管理，导致之前实例的关联被覆盖。移除循环内的提交，统一在所有实例添加完成后提交：

for pub in pubs:
    # ... 保留其他逻辑 ...
    pub_obj.ed_system = ed_system_objs[0]
    pub_obj.editor = editor_objs[randint(0, len(editor_objs) - 1)]
    session.add(pub_obj)
# 移除循环内的session.commit()，统一在此提交
try:
    session.commit()
except sqlalchemy.exc.IntegrityError as err:
    print("Initial registries already exist, SKIPPING!")
except Exception as err:
    print(f"{err}")

3. 优化EducationalSystem的提交逻辑 #

创建EducationalSystem时也无需循环内提交，批量添加后统一提交即可：

for ed_system in ed_systems:
    ed_system_obj = EducationalSystem(name=ed_system)
    ed_system_objs.append(ed_system_obj)
    session.add(ed_system_obj)
# 移到循环外提交
session.commit()

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