CODEWITHSUNDEEP
pythonpython-3.xsqlalchemypsycopg2
Justcurious
Justcurious

Reputation: 2064

How to persist data into a One-to-Many SELF referential with SQLAlchemy?

I'm trying to persist a One-To-Many self-referential relationship. My table looks something like this:

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

I arrived at this format following this page in the docs for one-to-many and another page describing how to declare self referential relationships. I've also already tried with the following variations:

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

I can insert the rows, query, commit, etc... but when trying to define a relationship, it fails with the following:

Example One:

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

Will raise:

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...

Example Two (with connected_with attr):

u1.connected_with = [u2.id]

Will Raise:

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

Example Three (with the object itself):

u1.connected_ids = [u2]

Will raise:

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

At this point, my best guess is that the table is not defined the way I expect it to, but I also don't know what is wrong in it.

Any pointers and help will be appreciated.

Upvotes: 0

Views: 237

Answers (2)

Louis Huang
Louis Huang

Reputation: 994

As of SQLAlchemy 2.0, to define a self-referential relationship:

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")

Using the model:

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()

Upvotes: 0

Ian Wilson
Ian Wilson

Reputation: 9089

I'm not sure if adjacency list is the pattern you want. Here is how it could work though. The ForeignKey should be a scalar, ie. single value, not a list. A many side of the relationship can be a list though, here it is 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)

Upvotes: 1

Related Questions