如何构造自引用/递归 SQLModel答案

【问题标题】：How do I construct a self-referential/recursive SQLModel如何构造自引用/递归 SQLModel
【发布时间】：2022-08-20 00:59:14
【问题描述】：

我想定义一个具有自我参照（或者递归的) 外键使用SQLModel。（这种关系模式有时也称为邻接表.) 纯SQLAlchemy 实现在他们的文档中描述为here。

假设我要实现基本的树状结构如上面链接的SQLAlchemy 示例中所述，其中我有一个Node 模型，每个实例都有一个id 主键、一个data 字段（例如str 类型）和一个可选参考（阅读外键) 到另一个我们称之为它的节点父母节点（字段名称parent_id）。

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

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

问题是双重的：

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

标签： python sqlalchemy adjacency-list self-referencing-table sqlmodel

【解决方案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)
)

【讨论】：