如何使用 SQLAlchemy 将数据持久化到一对多 SELF 引用中？

Question

我正在努力维持一对多自我参照关系。我的桌子看起来像这样：

class Users(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, unique=True)
    connected_ids = Column(Integer, ForeignKey("users.id"))
    connected_with = relationship("Users")

我在 一对多文档中的这一页和另一个 描述如何声明自引用关系的页面之后得到了这种格式。我也已经尝试过以下变体：

connected_with = relationship("Users", backref="users")
connected_with = relationship("Users", backref="users", remote_side="users.c.id"")

我可以插入行、查询、提交等...但是在尝试定义关系时，它失败了，并显示以下内容：

例一：

u1 = session.get(Users, 1)
u2 = session.get(Users, 2)
u1.connected_ids = [u2.id]

会提高：

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.DatatypeMismatch) column "connected_ids" is of type integer but expression is of type integer[]
    LINE 1: ...users SET last_updated=now(), connected_ids=ARRAY[2911...

示例二（带有connected_with attr）：

u1.connected_with = [u2.id]

会提高：

AttributeError: 'int' object has no attribute '_sa_instance_state'

示例三（对象本身）：

u1.connected_ids = [u2]

会提高：

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) can't adapt type 'Users'

在这一点上，我最好的猜测是表格没有按照我期望的方式定义，但我也不知道它有什么问题。

任何指点和帮助将不胜感激。

Answer 1

我不确定邻接表是否是你想要的模式。这是它如何工作的。 ForeignKey 应该是标量，即。单个值，而不是列表。关系的

many

面可以是一个列表，这里是

children

:

import sys

from sqlalchemy import (
    create_engine,
    Integer,
    String,
    ForeignKey,
)
from sqlalchemy.schema import (
    Column,
)
from sqlalchemy.orm import Session, declarative_base, relationship, backref


username, password, db = sys.argv[1:4]


Base = declarative_base()

engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=True)


metadata = Base.metadata


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)

    parent_id = Column(Integer, ForeignKey("users.id"), nullable=True)

    # OR children = relationship("User", backref=backref("parent", remote_side=id))
    parent = relationship("User", back_populates="children", remote_side=id)
    children = relationship("User", back_populates="parent", remote_side=parent_id)


metadata.create_all(engine)


with Session(engine) as session, session.begin():
    # Create 3 users, 2 connected to root.
    root = User(name="root")
    a = User(name="a", parent=root)
    b = User(name="b", parent=root)
    session.add_all([root, a, b])


with Session(engine) as session, session.begin():
    # Check that root exists and both children are connected to it.
    root = session.query(User).where(User.name == "root").first()
    assert len(root.children) == 2
    for child in root.children:
        assert child.parent == root

with Session(engine) as session, session.begin():
    # Add another child to root using the children property with append.
    root = session.query(User).where(User.name == "root").first()
    root.children.append(User(name="c"))


with Session(engine) as session, session.begin():
    # Check that root exists and that there are now 3 children instead of 2.
    root = session.query(User).where(User.name == "root").first()
    print(root.name)
    assert len(root.children) == 3
    for child in root.children:
        assert child.parent == root
        print(child.name)

Answer 2

从 SQLAlchemy 2.0 开始，定义自引用关系：

pk_int = Annotated[int, mapped_column(primary_key=True)]
str255 = Annotated[str, 255]

class Base(DeclarativeBase):
    type_annotation_map = {
        str255: String(255),
    }
    
class Employee(Base):
    __tablename__ = 'employees'

    id: Mapped[pk_int]
    manager_id: Mapped[Optional[int]] = mapped_column(ForeignKey('employees.id'))
    name: Mapped[str255] = mapped_column(nullable=False)

    manager: Mapped["Employee"] = relationship(
        back_populates="reports", remote_side="Employee.id")
    reports: Mapped[list["Employee"]] = relationship(
        back_populates="manager", remote_side="Employee.manager_id")

使用模型：

louis = Employee(name="Louis")
alice = Employee(name="alice")
bob = Employee(name="bob")
louis.reports.append(alice)
louis.reports.append(bob)
session.add(louis)

session.commit()