<p dir="auto"><strong>Migrated issue, originally created by Sheer El Showk (<a href="https://github.com/sheer">@sheer</a>)</strong></p>
<p dir="auto">This bug was discussed on stack overview here:</p>
<p dir="auto">[http://stackoverflow.com/questions/33888539/getting-sqlalchemy-to-do-on-duplicate-key-update-inside-an-orm-cascade-in-mys?noredirect=1#comment55890922_33888539<br>
](Link URL)</p>
<p dir="auto">We create a simple object hierarchy: Groups contain Users and Users have Email addresses.  We want the email address to be stored uniquely even if its shared between users.  Constructing two users with the same address and using session.merge() to add them has the correct behaviour (the same key is reused and no error is thrown).  If, on the other hand, we add the two users to a group and then use session.merge() on the group instead them the two identical addresses lead to a unique key exception on address (due to an insert many).</p>
<p dir="auto">Here is the relevant code:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from sqlalchemy import create_engine, Column, types
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.orm import Session

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref


engine = create_engine('sqlite:///:memory:', echo=False)
Base = declarative_base()

session = scoped_session(sessionmaker(bind=engine))

class Group(Base):
    __tablename__ = &quot;groups&quot;
    gid = Column(types.Integer, primary_key=True)
    name = Column(types.String(255))

    users = relationship(&quot;User&quot;, backref=&quot;group&quot;)

    def __repr__(self):
        ret =  &quot;Group(name=%r)&quot; % self.name
        for user in self.users:
            ret += str(user)

class User(Base):
    __tablename__ = &quot;users&quot;
    login = Column(types.String(50), primary_key=True)
    name = Column(types.String(255))
    group_id = Column(types.Integer, ForeignKey('groups.gid'))
    address = Column(types.String(200), 
                            ForeignKey('addresses.email_address'))
    email = relationship(&quot;Address&quot;)

    def __repr__(self):
        return &quot;User(login=%r, name=%r)\n%s&quot; % (self.login, self.name,
                str(self.email))


class Address(Base):
    __tablename__ = 'addresses'
    email_address = Column(types.String(200), nullable=False, primary_key=True)
    #user_login = Column(types.String(50), ForeignKey('users.login'))

    def __repr__(self):
        return &quot;&lt;Address(email_address='%s')&gt;&quot; % self.email_address


Base.metadata.create_all(engine)

if __name__ == '__main__':

    # this works correctly even though we reuse a unique key
    u1 = User(login='Guy', name=&quot;Some Guy&quot;)
    u1.email=Address(email_address='nameless@yahoo.com')
    u2 = User(login='Gal', name=&quot;Some Gal&quot;)
    u2.email=Address(email_address='nameless@yahoo.com')
    session.merge(u1) 
    session.merge(u2) 
    session.commit()

    print(&quot;two users with addresses&quot;)
    for u in session.query(User):
        print(u)


    # though this is similar it ends up using insertmany and throws a unique key
    # constraint even with the merge
    u3 = User(login='Mr. User', name=&quot;A dude&quot;)
    u3.email=Address(email_address='james@yahoo.com')
    u4 = User(login='Mrs. User', name=&quot;A dudette&quot;)
    u4.email=Address(email_address='jill@yahoo.com')
    u5 = User(login='Mrs. User2', name=&quot;A dudette2&quot;)
    u5.email=Address(email_address='jill@yahoo.com')
    g1 = Group(name=&quot;G1&quot;)
    g1.users.append(u3)
    g1.users.append(u4)
    g1.users.append(u5)
    session.merge(g1) 
    session.commit()

    print(g1)"><pre class="notranslate"><code class="notranslate">from sqlalchemy import create_engine, Column, types
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.orm import Session

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref


engine = create_engine('sqlite:///:memory:', echo=False)
Base = declarative_base()

session = scoped_session(sessionmaker(bind=engine))

class Group(Base):
    __tablename__ = "groups"
    gid = Column(types.Integer, primary_key=True)
    name = Column(types.String(255))

    users = relationship("User", backref="group")

    def __repr__(self):
        ret =  "Group(name=%r)" % self.name
        for user in self.users:
            ret += str(user)

class User(Base):
    __tablename__ = "users"
    login = Column(types.String(50), primary_key=True)
    name = Column(types.String(255))
    group_id = Column(types.Integer, ForeignKey('groups.gid'))
    address = Column(types.String(200), 
                            ForeignKey('addresses.email_address'))
    email = relationship("Address")

    def __repr__(self):
        return "User(login=%r, name=%r)\n%s" % (self.login, self.name,
                str(self.email))


class Address(Base):
    __tablename__ = 'addresses'
    email_address = Column(types.String(200), nullable=False, primary_key=True)
    #user_login = Column(types.String(50), ForeignKey('users.login'))

    def __repr__(self):
        return "&lt;Address(email_address='%s')&gt;" % self.email_address


Base.metadata.create_all(engine)

if __name__ == '__main__':

    # this works correctly even though we reuse a unique key
    u1 = User(login='Guy', name="Some Guy")
    u1.email=Address(email_address='nameless@yahoo.com')
    u2 = User(login='Gal', name="Some Gal")
    u2.email=Address(email_address='nameless@yahoo.com')
    session.merge(u1) 
    session.merge(u2) 
    session.commit()

    print("two users with addresses")
    for u in session.query(User):
        print(u)


    # though this is similar it ends up using insertmany and throws a unique key
    # constraint even with the merge
    u3 = User(login='Mr. User', name="A dude")
    u3.email=Address(email_address='james@yahoo.com')
    u4 = User(login='Mrs. User', name="A dudette")
    u4.email=Address(email_address='jill@yahoo.com')
    u5 = User(login='Mrs. User2', name="A dudette2")
    u5.email=Address(email_address='jill@yahoo.com')
    g1 = Group(name="G1")
    g1.users.append(u3)
    g1.users.append(u4)
    g1.users.append(u5)
    session.merge(g1) 
    session.commit()

    print(g1)
</code></pre></div>
<p dir="auto">Replace the asyncpg paramstyle from format to numeric, using <code class="notranslate">$</code> instead of the standard <code class="notranslate">:</code>.</p>
<p dir="auto">The main advantage of this style is that allows repeated parameters to not be duplicated</p>