Skip to content

Instantly share code, notes, and snippets.

@funseiki
Last active August 12, 2019 22:47
Show Gist options
  • Save funseiki/bd28e24d6b7cf583c993a45fd0a504c8 to your computer and use it in GitHub Desktop.
Save funseiki/bd28e24d6b7cf583c993a45fd0a504c8 to your computer and use it in GitHub Desktop.
SqlAlchemy Eager Loading Example Issue
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
def buildConnectionString(db_type, server, port, name, user, password):
# We're using an in-memory data
if db_type == "sqlite":
return "sqlite://"
authStr = user + ":" + password + "@" if user and len(user) and len(password) else ""
# E.g. postgresql://localhost:5432/my_db
return "{}://{}{}:{}/{}".format(db_type, authStr, server, port, name)
connectionString = buildConnectionString(\
"postgresql",
"localhost",
"5432",
"",
"",
None)
Base = declarative_base()
class Thing(Base):
__tablename__ = "Things"
id = Column(Integer, primary_key=True)
name = Column(String(256))
thingType = Column(String(256))
__mapper_args__ = {
'with_polymorphic':'*',
'polymorphic_load':'inline',
'polymorphic_on':"thingType",
'polymorphic_identity':"thing"
}
def __init__(self, **kwds):
for key in kwds:
setattr(self, key, kwds[key])
class Ball(Thing):
__tablename__ = "Balls"
id = Column('id', Integer, ForeignKey('Things.id'), primary_key=True)
color = Column('color', String(256))
__mapper_args__ = {
'polymorphic_identity':'ball'
}
pass
class Bin(Thing):
__tablename__ = "Bins"
id = Column('id', Integer, ForeignKey('Things.id'), primary_key=True)
ballId = Column('ballId', Integer, ForeignKey('Balls.id'))
ball = relationship(Ball, foreign_keys=[ballId], backref="outputBins")
__mapper_args__ = {
'polymorphic_identity':'bin'
}
pass
class Court(Thing):
__tablename__ = "Courts"
id = Column('id', Integer, ForeignKey('Things.id'), primary_key=True)
homeBinId = Column('homeBinId', Integer, ForeignKey('Bins.id'))
awayBinId = Column('awayBinId', Integer, ForeignKey('Bins.id'))
homeBin = relationship(Bin, foreign_keys=[homeBinId], backref="homeCourts")
awayBin = relationship(Bin, foreign_keys=[awayBinId], backref="awayCourts")
__mapper_args__ = {
'polymorphic_identity':'court'
}
def buildTables(metadata):
thingTable = Table(
'Things', metadata,
Column('id', Integer(), primary_key=True),
Column('thingType', String(256)),
Column('name', String(256))
)
ballTable = Table(
'Balls', metadata,
Column('id', Integer, ForeignKey(thingTable.c.id), primary_key=True),
Column('color', String(256))
)
binTable = Table(
'Bins', metadata,
Column('id', Integer, ForeignKey(thingTable.c.id), primary_key=True),
Column('ballId', Integer, ForeignKey(ballTable.c.id)),
)
courtTable = Table(
'Courts', metadata,
Column('id', Integer, ForeignKey(thingTable.c.id), primary_key=True),
Column('homeBinId', Integer, ForeignKey(binTable.c.id)),
Column('awayBinId', Integer, ForeignKey(binTable.c.id)),
)
return
metadata = MetaData()
engine = create_engine(connectionString)
buildTables(metadata)
Session = sessionmaker(bind=engine)
session = Session()
def initializeWithData():
baseball = Ball(name="baseball", thingType="ball", color="white")
bin = Bin(name="bin", thingType="bin", ball=baseball)
court = Court(name="SomeCourt", thingType="court", homeBin=bin)
session.add(baseball)
session.add(bin)
session.commit()
return
# Uncomment the below to create the tables in the database and add some reason
#metadata.create_all(engine)
#initializeWithData()
def courtQuery():
awayBalls = aliased(Ball, name="awayBalls")
homeBalls = aliased(Ball, name="homeBalls")
awayBins = aliased(Bin, name="awayBins")
homeBins = aliased(Bin, name="homeBins")
query = session.query(Court)\
.outerjoin(awayBins, Court.awayBinId == awayBins.id)\
.outerjoin(awayBalls, awayBins.ballId == awayBalls.id)\
.outerjoin(homeBins, Court.homeBinId == homeBins.id)\
.outerjoin(homeBalls, homeBins.ballId == homeBalls.id)\
.options(contains_eager(Court.awayBin, alias=awayBins).contains_eager(awayBins.ball, alias=awayBalls))\
.options(contains_eager(Court.homeBin, alias=homeBins).contains_eager(homeBins.ball, alias=homeBalls))
return [r for r in query]
def checkData():
import time
timesChecked = 0
while(True):
print("Times checked = {}".format(timesChecked))
results = courtQuery()
court = results[0]
ball = court.homeBin.ball
ballColor = ball.color
print(ballColor)
print("Sleeping")
timesChecked += 1
time.sleep(5)
return
checkData()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment