Over the years, I heavily mutilated example.trajectory from its original and ended up with the following in my db.py module to be able to reuse a database session. My issue was that a new session would attempt to remap an already mapped SQLAlchemy Table object.
FWIW maybe you can use below as inspiration...
class ErpEdiemSession(object):
"""Reusing a single instance of the database session client
The session itself is not persisted, according to
https://github.com/zopefoundation/zope.sqlalchemy/tree/1.6
"""
# Enforce UTC as our base timezone. See UTCDateTime class below.
os.putenv("PGTZ", "UTC")
_instance = None
_mapped_classes = {}
def __new__(cls):
"""This code brought to you by ChatGPT
using a singleton pattern
"""
if cls._instance is None:
cls._instance = super(ErpEdiemSession, cls).__new__(cls)
initializeSession() # Initialize the session
# log.info('Initializing erpediem session object')
else:
# log.info('Not initializing session again')
pass
return cls._instance
def __call__(self, dbconnection):
"""Always returns the scoped session
<sqlalchemy.orm.scoping.scoped_session object at 0x7f9a708d5950>
so that we can use self.session.get_bind() in the view
"""
global _PROFILE_ENGINE
global _PROFILE_SESSION
global DEBUG
"""
if _PROFILE_SESSION:
log.info('Calling erpediem session object %s %s %s' % (self, dbconnection, _PROFILE_SESSION, ))
log.info(self._mapped_classes)
else:
log.info('no profile session yet')
"""
portal = api.portal.get()
saconnect = ISQLAlchemyConnectionStrings(portal)
self.dsn = saconnect[dbconnection]
# log.info('erpediem profile engine %s' % _PROFILE_ENGINE)
# log.info('erpediem profile session %s' % _PROFILE_SESSION)
if not (_PROFILE_ENGINE and _PROFILE_SESSION):
'''
eleddy: since we moved this code out of the actual init, so that we can
have a sane configuration setup, we MUST do this with a lock. Otherwise sql
alchemy loses its shit and goes on a 'fuck your multi-threading - I'll eat
pancakes on your grave!' tirade. Then you spend your friday
night sober and staring at an abyss of configuration headaches and
infinite loops usually reserved for Mondays.
'''
with _DB_CONFIG_LOCK:
if not _PROFILE_ENGINE:
_create_database_if_missing(self.dsn)
_PROFILE_ENGINE = create_engine(
self.dsn,
pool_size=5,
pool_recycle=3600,
convert_unicode=True,
echo=DEBUG,
)
# reflect existing tables
Base.metadata.reflect(_PROFILE_ENGINE)
# create new tables
Base.metadata.create_all(_PROFILE_ENGINE)
"""
for _t in Base.metadata.tables:
columns = Table(_t, Base.metadata).columns
log.info("Table: %s" % columns)
"""
if not _PROFILE_SESSION:
_PROFILE_SESSION = scoped_session(sessionmaker(bind=_PROFILE_ENGINE))
return _PROFILE_SESSION
def map_class(self, class_to_map, table, primary_key):
"""This code brought to you by ChatGPT
using a registry of mapped classes
"""
# log.info('Mapped classes: %s' % self._mapped_classes)
if class_to_map not in self._mapped_classes.keys():
# log.info('Attempting to map: %s' % class_to_map)
mapper(class_to_map, table, primary_key=primary_key)
self._mapped_classes[class_to_map] = True
else:
# log.info('Not mapping: %s again' % class_to_map)
pass