如何构建自引用/递归 SQLModel

Question

我想定义一个具有自引用（或递归）外键的模型，使用SQLModel. （这种关系模式有时也称为邻接列表。）纯实现在SQLAlchemy其文档中进行了描述。

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

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

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

问题是双重的：

1. 实现这个的优雅方法是什么SQLModel？

2. 我如何创建这样的节点实例并将它们插入数据库？

Answer 1

该sqlmodel.Relationship函数允许显式地将附加关键字参数传递给sqlalchemy.orm.relationship通过参数在幕后调用的构造函数sa_relationship_kwargs。该参数需要一个表示参数名称的字符串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以避免当我们想要创建一个实例时被 IDE 打扰，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)
)