[Sqlalchemy-tickets] Issue #4040: expunge pending orphans on flush that weren't caught by other means (zzzeek/sqlalchemy)

[Michael B. <iss...@bi...>]

New issue 4040: expunge pending orphans on flush that weren't caught by other means
https://bitbucket.org/zzzeek/sqlalchemy/issues/4040/expunge-pending-orphans-on-flush-that

Michael Bayer:

```
#!python

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class A1(Base):
    __tablename__ = 'a1'
    id = Column(Integer, primary_key=True)
    bs = relationship("B", cascade="all, delete-orphan")


class A2(Base):
    __tablename__ = 'a2'
    id = Column(Integer, primary_key=True)
    bs = relationship("B", cascade="all, delete-orphan")


class B(Base):
    __tablename__ = 'b'
    id = Column(Integer, primary_key=True)
    a1_id = Column(ForeignKey('a1.id'))
    a2_id = Column(ForeignKey('a2.id'))

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)

b_orphan1, b_orphan2 = B(), B()

a1a, a1b = A1(), A1()

a2a = A2()

a2a.bs.append(b_orphan1)
a2a.bs.append(b_orphan2)

s.add(a2a)
s.add(a1a)

# add it here, it works
# s.add(a1b)

a1a.bs.append(b_orphan1)
a1b.bs.append(b_orphan2)

a1a.bs.remove(b_orphan1)
a1b.bs.remove(b_orphan2)

# down here, fails
s.add(a1b)

s.commit()

assert len(s.query(B).all()) == 0


```

fix is to check for pending orphans once more at the same time we check for persistent orphans:


```
#!diff



diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 7c313e6..bade700 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -2271,11 +2271,13 @@ class Session(_SessionClassMethods):
             proc = new.union(dirty).difference(deleted)
 
         for state in proc:
-            is_orphan = (
-                _state_mapper(state)._is_orphan(state) and state.has_identity)
-            _reg = flush_context.register_object(state, isdelete=is_orphan)
-            assert _reg, "Failed to add object to the flush context!"
-            processed.add(state)
+            is_orphan = _state_mapper(state)._is_orphan(state)
+            if is_orphan and not state.has_identity:
+                self._expunge_states([state])
+            else:
+                _reg = flush_context.register_object(state, isdelete=is_orphan)
+                assert _reg, "Failed to add object to the flush context!"
+                processed.add(state)
 
         # put all remaining deletes into the flush context.
         if objset:

```