SQLAlchemy+SQLModel中嵌套关系的递归遍历与更新问题

我正在用Python的SQLAlchemy ORM结合SQLModel处理多级关联模型，包含多对一、多对多关系，目标是在创建完整对象图后遍历所有关系，判断子关系是否发生变化，从而决定是更新现有对象（比如last_purchased_at字段）还是创建新对象。所有模型都带有唯一约束，数据会从多个数据源获取，构建完对象图后再做持久化。

示例模型 #

# Many-to-Many relationship table between Book and Genre
book_genre_table = Table('book_genre', Base.metadata,
    Column('book_id', Integer, ForeignKey('books.id')),
    Column('genre_id', Integer, ForeignKey('genres.id'))
)

class Author(Base):
    __tablename__ = 'authors'
    __table_args__ = (UniqueConstraint('name'),)

    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)  # Unique constraint on the 'name' field

    # One-to-Many relationship with Book
    books = relationship('Book', back_populates='author')

class Book(Base):
    __tablename__ = 'books'

    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))
    last_purchased_at = Column(DateTime)

    # Many-to-One relationship with Author
    author = relationship('Author', back_populates='books')

    # Many-to-Many relationship with Genre
    genres = relationship('Genre', secondary=book_genre_table, back_populates='books')

class Genre(Base):
    __tablename__ = 'genres'
    __table_args__ = (UniqueConstraint('name'),)

    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)  # Unique constraint on the 'name' field

    # Many-to-Many relationship with Book
    books = relationship('Book', secondary=book_genre_table, back_populates='books')

核心挑战 #

• 关系变更识别：新增书籍或更新现有书籍详情（包括关联流派的变更）时，需要准确识别这些变化。
• 唯一约束处理：Author.name和Genre.name带有唯一约束，需处理新书关联已有作者或流派的场景。
• 效率问题：随着关系深度增加，遍历和比较嵌套对象的效率令人担忧。

目前没有找到能递归遍历SQLAlchemy对象图的方案，无法统一处理以下场景：1）判断对象是否存在，存在则关联已有对象；2）判断对象是否变更，需更新；3）不存在则创建新对象。

现有尝试方案 #

from typing import Type, Any, Optional, Dict
from sqlalchemy.future import select
from sqlmodel import Session, SQLModel, Relationship
from sqlalchemy.orm import RelationshipProperty
import pydantic
from sqlalchemy.orm import selectinload
import typing
import sqlalchemy


def find_relationships(model: Type[SQLModel]) -> typing.List[str]:
    relationships = []
    for attr_name, attr_type in model.__annotations__.items():
        # Check if attr_type is a generic type (e.g., List, Optional)
        if isinstance(attr_type, typing._GenericAlias):
            # Check type arguments of the generic type
            for arg in typing.get_args(attr_type):
                if isinstance(arg, typing.ForwardRef) or (isinstance(arg, type) and issubclass(arg, SQLModel)):
                    relationships.append(attr_name)
                    break  # Found a relationship, no need to check other args
        elif isinstance(attr_type, type) and issubclass(attr_type, SQLModel):
            relationships.append(attr_name)

    return relationships


async def load_relationships(async_session,
                             model_instance,
                             relationship_keys):
    options = [selectinload(getattr(model_instance.__class__, attr)) for attr in relationship_keys]
    stmt = select(model_instance.__class__).options(*options).where(model_instance.__class__.id == model_instance.id)
    result = await async_session.execute(stmt)
    return result.scalars().first()


async def find_existing_instance(async_session, model, **kwargs):
    """
    Find an existing instance of a model based on provided attributes (kwargs).
    If found, return the instance; otherwise, return None.
    """
    stmt = select(model).where(
        *[getattr(model, key) == value for key, value in kwargs.items()]
    )
    result = await async_session.execute(stmt)
    instance = result.scalars().first()
    return instance


def find_unique_constraints(model: Type[SQLModel]) -> typing.List[typing.Union[sqlalchemy.UniqueConstraint, sqlalchemy.Column]]:
    unique_constraints = set()

    # Check for table-level unique constraints in __table_args__
    if hasattr(model, '__table_args__'):
        for arg in model.__table_args__:
            if isinstance(arg, sqlalchemy.UniqueConstraint):
                for column in arg:
                    unique_constraints.add(column.name)

    # Check for column-level unique constraints
    for _, attr in model.__dict__.items():
        if isinstance(attr, sqlalchemy.Column) and attr.unique:
            unique_constraints.add(attr)

    return unique_constraints


async def find_or_create(async_session, new_instance, recurse: bool = True):
    unique_constraint_keys = find_unique_constraints(new_instance.__class__)
    print(f"model: {new_instance.__class__}, unique_constraint_keys: {unique_constraint_keys}")
    search_fields = new_instance.model_dump(include=unique_constraint_keys)

    # Check for existing instance of the class based on unique constraints
    existing_instance = await find_existing_instance(async_session, new_instance.__class__, **search_fields)
    if existing_instance:
        print(f"Found existing instance: {existing_instance.id} for {existing_instance.__class__}")
        if recurse:

            # If found, load all relationships
            relationship_keys = find_relationships(new_instance.__class__)
            new_instance = await load_relationships(async_session, new_instance, relationship_keys)
            print(f"New instance is {new_instance.model_dump()}")
            existing_instance = await load_relationships(async_session, existing_instance, relationship_keys)
            print(f"Existing instance is {existing_instance.model_dump()}")
            for key in relationship_keys:
                new_attribute = getattr(new_instance, key, None)
                existing_attribute = getattr(existing_instance, key, None)

                # Check if new and existing attribute are lists
                if isinstance(new_attribute, list) and isinstance(existing_attribute, list):
                    if len(new_attribute) == 0 and len(existing_attribute) == 0:
                        continue
                    if len(new_attribute) > 0:
                        new_attribute_unique_constraint_keys = find_unique_constraints(new_attribute[0].__class__)
                        new_attribute_dicts = [item.model_dump(include=new_attribute_unique_constraint_keys) for item in new_attribute]
                    if len(existing_attribute) > 0:
                        existing_attribute_unique_constraint_keys = find_unique_constraints(existing_attribute[0].__class__)
                        existing_attribute_dicts = [item.model_dump(include=existing_attribute_unique_constraint_keys) for item in existing_attribute]

                    # Check if new attribute is a subset of existing attribute
                    for item_dict, item in zip(new_attribute_dicts, new_attribute):
                        if item_dict not in existing_attribute_dicts:
                            await find_or_create(async_session, item)
                else:
                    new_attribute_unique_constraint_keys = find_unique_constraints(new_attribute.__class__)
                    new_attribute_dict = new_attribute.model_dump(include=new_attribute_unique_constraint_keys)
                    existing_attribute_unique_constraint_keys = find_unique_constraints(existing_attribute.__class__)
                    existing_attribute_dict = existing_attribute.model_dump(include=existing_attribute_unique_constraint_keys)
                    if new_attribute_dict != existing_attribute_dict:
                        await find_or_create(async_session, new_attribute)

        await async_session.commit()
        await async_session.refresh(existing_instance)
        return existing_instance

    async_session.add(new_instance)
    await async_session.commit()
    await async_session.refresh(new_instance)

    if recurse:
        relationship_keys = find_relationships(new_instance.__class__)
        new_instance = await load_relationships(async_session, new_instance, relationship_keys)
        for key in relationship_keys:
            new_attribute = getattr(new_instance, key, None)
            if isinstance(new_attribute, list):
                for item in new_attribute:
                    await find_or_create(async_session, item)
            else:
                await find_or_create(async_session, new_attribute)

    return new_instance

当前方案存在的问题 #

• 关系加载效率低下：每次处理关系都要单独查询加载，随着关系层级增多，查询次数会指数级增长。
• 异步会话提交问题：在递归过程中频繁调用await async_session.commit()，会导致事务粒度碎片化，不仅影响效率，还可能引发一致性问题。

希望获得针对这些问题的解决方案，或是相关的实现思路参考。

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