from__future__importannotationsimportloggingfromtypingimportTYPE_CHECKING,Dict,Tupleimportosimportjsonpickleimportjsonpickle.ext.numpyasjsonpickle_numpyimportsqlalchemy.excimportsqlalchemy.ormfrompalaestrai.core.runtime_configimportRuntimeConfigifTYPE_CHECKING:importsqlalchemy.engineimportsqlalchemy.orm.sessionLOG=logging.getLogger(__name__)_session_makers:Dict[int,Tuple[sqlalchemy.engine.Engine,sqlalchemy.orm.session.sessionmaker]]={}def_get_session_maker()->sqlalchemy.orm.session.sessionmaker:"""Internal getter ensure one session maker per process"""pid=os.getpid()if(pidin_session_makersandstr(_session_makers[pid][0].url)==RuntimeConfig().store_uri):LOG.debug("Returning existing DB session maker for PID %d: %s",pid,_session_makers[pid][1],)return_session_makers[pid][1]jsonpickle_numpy.register_handlers()jsonpickle.set_preferred_backend("simplejson")jsonpickle.set_encoder_options("simplejson",ignore_nan=True)db_engine=sqlalchemy.create_engine(RuntimeConfig().store_uri,json_serializer=jsonpickle.dumps,json_deserializer=jsonpickle.loads,)db_session_maker=sqlalchemy.orm.sessionmaker()db_session_maker.configure(bind=db_engine)_session_makers[pid]=(db_engine,db_session_maker)LOG.debug("Created new DB session maker for PID %d; state dict is now: %s",pid,_session_makers,)returndb_session_maker
[docs]defSession()->sqlalchemy.orm.Session:"""Creates a new, connected database session to run queries on. This is a convenience function that creates and returns a new, opened database session. It uses the access data provided by ::`RuntimeConfig.store_uri`. It exists in order to facilitate working with the store, such as this:: from palaestrai.store import Session, database_model as dbm session = Session() q = session.query(dbm.Experiment) Returns: -------- sqlalchemy.orm.Session The initialized, opened database session. """db_session_maker=_get_session_maker()returndb_session_maker()