sqla-merge-integrityerror.py

from sqlalchemy import create_engine, Column, String
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import sessionmaker, scoped_session

engine = create_engine(
    'mysql://root@mysql/airflow',
    echo=True
)

db_session = scoped_session(
    sessionmaker(autocommit=False, bind=engine)
)


class Base:
    @declared_attr
    def __tablename__(cls):
        return (
            "".join("_%s" % c if c.isupper() else c for c in cls.__name__)
            .strip("_")
            .lower()
        )


Base = declarative_base(cls=Base)
class MultiPK(Base):
    """test"""

    code = Column(String(50), primary_key=True, autoincrement=False)
    other = Column(String(50), primary_key=True, autoincrement=False)
    data = Column(String(50))


Base.metadata.create_all(engine)


db_session.add(MultiPK(code="a", other="b", data="pre"))
db_session.commit()

db_session.close()
db_session = None


# At a later date, go and update it. I.e. this would happen in a different requres
db_session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)

db_session.merge(MultiPK(code="a", other="b", data="post"))
db_session.commit()