How do I use SQLAlchemy to cascade deletes in SQLite?

Question

I've been reading various examples from SQLAlchemy documentation for cascade deletes, but nothing I try seems to work. Below is some sample code adapted from that documentation, but using back_populates instead of backref, as I understand that backref is being deprecated.

In the "main" section below, I would expect that deleting the order that "contains" the items would delete the items as well, but that does not happen. Obviously I don't understand something about how to configure these tables... what is it?

# third party imports
from sqlalchemy import Column, ForeignKey, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy_utils import create_database, database_exists

Base = declarative_base()


class Order(Base):
    __tablename__ = "business_order"
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    items = relationship(
        "Item", back_populates="order", cascade="all, delete, delete-orphan"
    )


class Item(Base):
    __tablename__ = "business_item"
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    order_id = Column(Integer, ForeignKey("business_order.id"))
    order = relationship("Order", back_populates="items")


def get_session(url="sqlite:///:memory:", create_db=True):
    """Get a SQLAlchemy Session instance for input database URL.
    :param url:
      SQLAlchemy URL for database, described here:
        http://docs.sqlalchemy.org/en/latest/core/engines.html#database-urls.
    :param create_db:
      Boolean indicating whether to create database from scratch.
    :returns:
      Sqlalchemy Session instance.
    """
    # Create a sqlite in-memory database engine
    if not database_exists(url):
        if create_db:
            create_database(url)
        else:
            msg = (
                "Database does not exist, will not create without "
                "create_db turned on."
            )
            print(msg)
            return None

    connect_args = {}

    engine = create_engine(url, echo=False, connect_args=connect_args)
    Base.metadata.create_all(engine)

    # create a session object that we can use to insert and
    # extract information from the database
    Session = sessionmaker(bind=engine, autoflush=False)
    session = Session()

    return session


if __name__ == "__main__":
    sqlite_url = "sqlite:///test_sqlite.db"
    session = get_session(sqlite_url)
    order = Order(name="order1")
    session.add(order)
    item = Item(order_id=order.id, name="item1")
    session.add(item)
    session.commit()
    session.delete(order)  # should delete items too, right?
    session.commit()
    orders = session.query(Order).all()
    print(len(orders))  # this returns 0 as expected
    items = session.query(Item).all()
    print(len(items))  # this returns 1, why?

user_2144390 · Answer

Order has an (implicit) autoincrement PK. When you do

    order = Order(name="order1")
    session.add(order)

order.id is None. Therefore, when you do

   item = Item(order_id=order.id, name="item1")

item.order_id will also be None, so item is actually not associated with order. Therefore, the delete doesn't cascade.

order doesn't get its id until .flush() (or .commit()) is called. So you could either do

    order = Order(name="order1")
    session.add(order)
    session.flush()  # !
    item = Item(order_id=order.id, name="item1")
    session.add(item)
    session.commit()

or do

    order = Order(name="order1", items=[Item(name="item1")])
    session.add(order)
    session.commit()
    session.delete(order)  # should delete items too, right?
    session.commit()
    orders = session.query(Order).all()
    print(len(orders))  # this returns 0 as expected
    items = session.query(Item).all()
    print(len(items))  # this also returns 0 as expected

How do I use SQLAlchemy to cascade deletes in SQLite?

1 Answers