SQL Alchemy Many To Many Relation delete cascading

model.py

# domain model
class Company(base):
    id: int = Column(BigInteger, primary_key=True, autoincrement=True)
    addresses = relationship("CompanyAddress", backref=backref("company",cascade='all,delete-orphan'))

class User(base):
    id: int = Column(BigInteger, primary_key=True, autoincrement=True)
    addresses = relationship("UserAddress", backref=backref("user",cascade='all,delete-orphan'))

class Address(base):
    id: int = Column(BigInteger, primary_key=True, autoincrement=True)
    street: str = Column(String(255))
    zipcode: str = Column(String(10))
    city: str = Column(String(255))

class CompanyAddress(base):
    address_id = Column(BigInteger, ForeignKey(Address.id), primary_key=True)
    company_id = Column(BigInteger, ForeignKey(Company.id), primary_key=True)
    primary: bool = Column(Boolean, comment="primary address")
    address = relationship(Address)

class UserAddress(base):
    address_id = Column(BigInteger, ForeignKey(Address.id), primary_key=True)
    user_id = Column(BigInteger, ForeignKey(User.id), primary_key=True)
    primary: bool = Column(Boolean, comment="primary address")
    address = relationship(Address)

# data setup
a1 = Address(zipcode="22305", city="Hamburg")  # id: 1
a2 = Address(zipcode="88998", city="Munich")   # id: 2

c = Company()                                  # id: 100
c.addresses.append(CompanyAddress(address=a1))

u = User()                                     # id: 1000
u.addresses.append(UserAddress(address=a1))
u.addresses.append(UserAddress(address=a2))

#saving works, all addresses are attached correctly

# delete relation
u.addresses = [a1]

# sqlalchemy tries to insert a new address with the same id (1) and fails bc it still exists

# i want to break up the relation between User and Address but dont want to delete the address itself

I don't quite get what you are trying to achieve. User.addresses is foreign-keyed to UserAddress rather than Address, so assigning [a1] to u.adresses is fundamentally incorrect.

If you want to delete the relationship between u and a2, you should delete the corresponding UserAddress. Haven't tested it but I'd try:

u = User()
ua1 = UserAddress(address=a1)
ua2 = UserAddress(address=a2)

u.addresses.append(ua1)
u.addresses.append(ua2)

#saving works, all addresses are attached correctly

# delete relation
session.delete(ua2)

Well i think Change detection should notice the missing element on the collection and remove it by automatically?! Xucong Zhan <[email protected]> schrieb am Do., 24. Dez. 2020, 16:32:

…

***@***.**** commented on this gist. ------------------------------ I don't quite get what you are trying to achieve. User.addresses is foreign-keyed to UserAddress rather than Address, so assigning [a1] to u.adresses is fundamentally incorrect. If you want to delete the relationship between u and a2, you should delete the corresponding UserAddress. Haven't tested it but I'd try: u = User()ua1 = UserAddress(address=a1)ua2 = UserAddress(address=a2) u.addresses.append(ua1)u.addresses.append(ua2) #saving works, all addresses are attached correctly # delete relationsession.delete(ua2) — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <https://gist.github.com/fb90c30404ad89bb3764925a653dcc6d#gistcomment-3572193>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACHVV3US2FNL7PETTITKQLSWNNI3ANCNFSM4VIJ3XHA> .

Well if you want to reassign a new collection, you should do u.addresses = [ua1] instead of u.addresses = [a1]. a1 is an instance of Address, not UserAddress.

And yes if you have declared cascade="delete-orphan" on User.address, ua2 will be removed from the db table automatically. BTW, cascade should be configured on the "one" side of the one-to-many relationship. Including it in the backref function is incorrect. Instead of:

addresses = relationship("CompanyAddress", backref=backref("company",cascade='all,delete-orphan'))

You should do:

addresses = relationship("CompanyAddress", backref=backref("company"), cascade='all,delete-orphan')

A slightly modified version of your snippet that should run as-is (BigInteger of sqlite will not auto-increment hence the change):

from sqlalchemy import (Boolean, Integer, Column, ForeignKey, String,
                        create_engine)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import backref, relationship
from sqlalchemy.orm.session import Session, sessionmaker


engine = create_engine("sqlite:///company.db")
SessionMaker = sessionmaker(bind=engine)
session: Session = SessionMaker()

Base = declarative_base()


class Company(Base):
    __tablename__ = 'company'
    id: int = Column(Integer, primary_key=True, autoincrement=True)
    addresses = relationship("CompanyAddress",
                             backref=backref("company"),
                             cascade='all,delete-orphan')


class User(Base):
    __tablename__ = 'user'
    id: int = Column(Integer, primary_key=True, autoincrement=True)
    addresses = relationship("UserAddress",
                             backref=backref("user"),
                             cascade='all,delete-orphan')


class Address(Base):
    __tablename__ = 'address'
    id: int = Column(Integer, primary_key=True, autoincrement=True)
    street: str = Column(String(255))
    zipcode: str = Column(String(10))
    city: str = Column(String(255))


class CompanyAddress(Base):
    __tablename__ = 'company_address'
    address_id = Column(Integer, ForeignKey(Address.id), primary_key=True)
    company_id = Column(Integer, ForeignKey(Company.id), primary_key=True)
    primary: bool = Column(Boolean, comment="primary address")
    address = relationship(Address)


class UserAddress(Base):
    __tablename__ = 'user_address'
    address_id = Column(Integer, ForeignKey(Address.id), primary_key=True)
    user_id = Column(Integer, ForeignKey(User.id), primary_key=True)
    primary: bool = Column(Boolean, comment="primary address", default=False)
    address = relationship(Address)

    def __repr__(self) -> str:
        return f'<UserAddress (User:{self.user_id}, Address:{self.address_id})>'


Base.metadata.create_all(engine)

# data setup
a1 = Address(zipcode="22305", city="Hamburg")  # id: 1
a2 = Address(zipcode="88998", city="Munich")   # id: 2
# c = Company()                                  # id: 100
# c.addresses.append(CompanyAddress(address=a1))

u = User()                                     # id: 1000
ua1 = UserAddress(address=a1)
ua2 = UserAddress(address=a2)
u.addresses.append(ua1)
u.addresses.append(ua2)

# or you can simply do:
# ua1 = UserAddress(user=u, address=a1)
# ua2 = UserAddress(user=u, address=a2)

# saving works, all addresses are attached correctly
session.add(u)
session.commit()
print(f'{u.addresses=}')

user_addresses = session.query(UserAddress).all()
print(f'{user_addresses=}')

# delete relation
u.addresses = [ua1]
session.add(u)
session.commit()
print(f'{u.addresses=}')

# inspect the user_address table
# ua2 is now deleted
user_addresses = session.query(UserAddress).all()
print(f'{user_addresses=}')

It will create a local sqlite db company.db that you can inspect and play around with.