Skip to content

Instantly share code, notes, and snippets.

@funseiki
Created November 19, 2020 23:09
Show Gist options
  • Save funseiki/a73424bebfb0d809e6d934f699a725bf to your computer and use it in GitHub Desktop.
Save funseiki/a73424bebfb0d809e6d934f699a725bf to your computer and use it in GitHub Desktop.
Script to reproduce SQLAlchemy error
import sqlalchemy as sa
from sqlalchemy import create_engine, inspect, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship, aliased, contains_eager, scoped_session
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class ObjectType(Base):
__tablename__ = "objecttypes"
id=Column(Integer, primary_key=True)
name=Column(String)
def __repr__(self):
return "ObjectType({}, id={})".format(self.name, self.id)
engine = create_engine('sqlite:///:memory:', echo=False)
Base.metadata.create_all(engine) # We'll call this again later
sessionFactory = sessionmaker(bind=engine, expire_on_commit=False)
scopedSessionFactory = scoped_session(sessionFactory)
def startScope():
return scopedSessionFactory()
def endScope():
scopedSessionFactory().close()
scopedSessionFactory.remove()
return
def addObjectTypes():
"""
Add in all the object types to the db without using a session
"""
values = ["('location')", "('locationinput')", "('gate')"]
q = """INSERT INTO objecttypes (name) VALUES {}""".format(",".join(values))
engine.execute(q)
return
def buildObjectTypes():
addObjectTypes()
session = startScope()
types = session.query(ObjectType).all()
endScope()
return dict([(objType.name, objType) for objType in types])
# Holds all the types
typeDict = buildObjectTypes()
class Thing(Base):
__tablename__ = "things"
id=Column(Integer, primary_key=True)
name=Column(String)
object_type_id=Column(Integer, ForeignKey('objecttypes.id'))
objectType=relationship(ObjectType)
version=Column(Integer, nullable=False)
timeCreated=Column(DateTime)
__mapper_args__ = {
'version_id_col': version,
'polymorphic_on':object_type_id,
'with_polymorphic':'*',
'polymorphic_load':'inline',
}
def __repr__(self):
return "{}, id={}, type={}, version={}".format(self.name, self.id, self.objectType, self.version)
class Location(Thing):
__tablename__ = "locations"
id=Column(Integer, ForeignKey('things.id'),primary_key=True)
__mapper_args__ = {
'polymorphic_identity': typeDict['location'].id
}
class LocationInput(Thing):
__tablename__ = "locationinputs"
id=Column(Integer, ForeignKey('things.id'),primary_key=True)
previousGateId=Column(Integer, ForeignKey('gates.id'))
__mapper_args__ = {
'polymorphic_identity': typeDict['locationinput'].id
}
class Gate(Thing):
__tablename__ = "gates"
id=Column(Integer, ForeignKey('things.id'),primary_key=True)
originId=Column(Integer, ForeignKey('locations.id'))
origin=relationship(Location, foreign_keys=[originId], backref="originGates")
originInputId=Column(Integer, ForeignKey('locationinputs.id'))
originInput=relationship(LocationInput, foreign_keys=[originInputId], backref="originInputGates")
destinationId=Column(Integer, ForeignKey('locations.id'))
destination=relationship(Location, foreign_keys=[destinationId], backref="destinationGates")
destinationInputId=Column(Integer, ForeignKey('locationinputs.id'))
destinationInput=relationship(LocationInput, foreign_keys=[destinationInputId], backref="destinationInputGates")
__mapper_args__ = {
'polymorphic_identity': typeDict['gate'].id
}
LocationInput.previousGate = relationship(Gate, foreign_keys=[LocationInput.previousGateId])
Base.metadata.create_all(engine)
def write(obj):
session = scopedSessionFactory()
return session.merge(obj)
def reproduceError():
tokyo = Location(name="tokyo", objectType=typeDict['location'])
tokyo = write(tokyo)
# If I clear out the current session here, the error will occur
scopedSessionFactory().close()
scopedSessionFactory.remove()
westGate = Gate(name="westGate", destination=tokyo, objectType=typeDict['gate'])
westGate = write(westGate)
luggage = LocationInput(name="luggage", objectType=typeDict['locationinput'])
luggage = write(luggage)
# This is the line where the error occurs
eastGate = Gate(name="eastGate", origin=tokyo, destinationInput=luggage, objectType=typeDict['gate'])
eastGate = write(eastGate)
print(eastGate)
return
reproduceError()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment