marshmallow-testcase.py

from sqlalchemy import Column, Integer, String, ForeignKey, MetaData, create_engine
from sqlalchemy.orm import sessionmaker, relationship, scoped_session
from sqlalchemy.ext.declarative import declarative_base
from marshmallow import fields
from marshmallow_sqlalchemy import ModelSchema


engine = create_engine("postgresql://elemoine@localhost/gis", echo=True)
metadata = MetaData(engine)

Base = declarative_base(metadata=metadata)

Session = scoped_session(sessionmaker(bind=engine))


#
# Define SQLAlchemy models
#


class Parent(Base):
    __tablename__ = "parent"
    id = Column(Integer, ForeignKey("child.id", ondelete="CASCADE"), primary_key=True)
    child = relationship("Child", lazy="joined")


class Child(Base):
    __tablename__ = "child"
    id = Column(Integer, primary_key=True)
    name = Column(String)


#
# Define Marshmallow model schemas
#


class ParentSchema(ModelSchema):
    class Meta:
        model = Parent
        sqla_session = Session

    child = fields.Nested("ChildSchema")


class ChildSchema(ModelSchema):
    class Meta:
        model = Child
        sqla_session = Session


#
# Create the tables in the database
#

metadata.drop_all(checkfirst=True)
metadata.create_all()


#
# Add objects to the database
#

parent = Parent()
child = Child()
child.id = 1
child.name = "name"
parent.child = child

Session.add(parent)
Session.commit()

#
# Load objects using marshmallow-sqalchemy and trigger the "FlushError: New instance <>
# with identity key () conflicts with persistent instance <>" error
#

Session.expunge_all()

parent = Session.query(Parent).one()

json_data = {"child": {"id": 1, "name": "new name"}}

with Session.no_autoflush:
    instance = ParentSchema().load(json_data)

Session.add(instance.data)

Session.commit()