Skip to content

Instantly share code, notes, and snippets.

@digitalkaoz

digitalkaoz/model.py

Last active December 25, 2020 00:42
Show Gist options
  • Save digitalkaoz/fb90c30404ad89bb3764925a653dcc6d to your computer and use it in GitHub Desktop.
Save digitalkaoz/fb90c30404ad89bb3764925a653dcc6d to your computer and use it in GitHub Desktop.
Download ZIP
SQL Alchemy Many To Many Relation delete cascading
Raw
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
@HymanZHAN
Copy link

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)

@digitalkaoz
Copy link
Author

digitalkaoz commented Dec 24, 2020 via email

@HymanZHAN
Copy link

HymanZHAN commented Dec 25, 2020 •
edited
Loading

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment