如何构造自引用/递归SQLModel？

Question

我想要定义一个使用self-referential (或递归)外键使用SQLModel的模型。(这种关系模式有时也称为邻接列表。)纯SQLAlchemy实现在其文档中被描述为这里。

假设我想实现上面链接的示例中描述的基本SQLAlchemy树结构，其中我有一个Node模型，每个实例都有一个id主键、一个data字段(例如str类型)，以及一个对另一个节点的可选引用(读取外键)，我们称之为它的父级节点(字段名parent_id)。

理想情况下，如果节点没有父节点，每个Node对象都应该有一个parent属性，即None；否则它将包含(指向)父Node对象的指针。

更好的是，每个Node对象都应该有一个children属性，它将是将其作为父对象引用的Node对象的列表。

问题有两个：

1. 用SQLModel实现这一点的优雅方法是什么？
2. 我将如何创建这样的节点实例并将它们插入数据库？

Answer 1

sqlmodel.Relationship函数允许显式地将附加关键字参数传递给通过sa_relationship_kwargs参数调用的sqlalchemy.orm.relationship构造函数。该参数需要将表示SQLAlchemy参数名称的字符串映射到我们希望作为参数传递的值。

因为SQLAlchemy关系在这种情况下提供了remote_side参数，所以我们可以直接利用它来用最少的代码构造自引用模式。文档在传递时提到了这一点，但关键是remote_side值。

在使用声明式时，可以作为Python可评估字符串传递。

这正是我们所需要的。因此，唯一缺失的部分是正确使用back_populates参数，我们可以这样构建模型：

from typing import Optional
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine


class Node(SQLModel, table=True):
    __tablename__ = 'node'  # just to be explicit

    id: Optional[int] = Field(default=None, primary_key=True)
    data: str
    parent_id: Optional[int] = Field(
        foreign_key='node.id',  # notice the lowercase "n" to refer to the database table name
        default=None,
        nullable=True
    )
    parent: Optional['Node'] = Relationship(
        back_populates='children',
        sa_relationship_kwargs=dict(
            remote_side='Node.id'  # notice the uppercase "N" to refer to this table class
        )
    )
    children: list['Node'] = Relationship(back_populates='parent')

# more code below...

侧注：我们将id定义为可选的，这是SQLModel的习惯做法，以避免在我们想要创建一个实例(只有在将id添加到数据库之后才知道它)时受到它的干扰。显然，parent_id和parent属性被定义为可选属性，因为并不是每个节点都需要在模型中有一个父节点。

要测试一切按我们预期的方式工作：

def test() -> None:
    # Initialize database & session:
    sqlite_file_name = 'database.db'
    sqlite_uri = f'sqlite:///{sqlite_file_name}'
    engine = create_engine(sqlite_uri, echo=True)
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)
    session = Session(engine)

    # Initialize nodes:
    root_node = Node(data='I am root')

    # Set the children's `parent` attributes;
    # the parent nodes' `children` lists are then set automatically:
    node_a = Node(parent=root_node, data='a')
    node_b = Node(parent=root_node, data='b')
    node_aa = Node(parent=node_a, data='aa')
    node_ab = Node(parent=node_a, data='ab')

    # Add to the parent node's `children` list;
    # the child node's `parent` attribute is then set automatically:
    node_ba = Node(data='ba')
    node_b.children.append(node_ba)

    # Commit to DB:
    session.add(root_node)
    session.commit()

    # Do some checks:
    assert root_node.children == [node_a, node_b]
    assert node_aa.parent.parent.children[1].parent is root_node
    assert node_ba.parent.data == 'b'
    assert all(n.data.startswith('a') for n in node_ab.parent.children)
    assert (node_ba.parent.parent.id == node_ba.parent.parent_id == root_node.id) \
           and isinstance(root_node.id, int)


if __name__ == '__main__':
    test()

所有的断言都是满意的，测试运行时没有任何问题。

此外，通过使用数据库引擎的echo=True开关，我们可以在日志输出中验证表的创建是否与我们预期的一样：

CREATE TABLE node (
    id INTEGER, 
    data VARCHAR NOT NULL, 
    parent_id INTEGER, 
    PRIMARY KEY (id), 
    FOREIGN KEY(parent_id) REFERENCES node (id)
)