Source code for libres.context.session

import re

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
from sqlalchemy.orm import scoped_session, sessionmaker

from libres.context.core import StoppableService


SERIALIZABLE = 'SERIALIZABLE'


[docs]class SessionProvider(StoppableService): """Global session utility. It provides a SERIALIZABLE session to libres. If you want to override this provider, be sure to set the isolation_level to SERIALIZABLE as well. If you don't do that, libres might run into errors as it assumes and tests against SERIALIZABLE connections! """
[docs] def __init__(self, dsn, engine_config={}, session_config={}): self.assert_valid_postgres_version(dsn) self.dsn = dsn self.engine = create_engine( dsn, poolclass=QueuePool, pool_size=5, max_overflow=5, isolation_level=SERIALIZABLE, **engine_config ) self.session = scoped_session(sessionmaker( bind=self.engine, **session_config ))
[docs] def stop_service(self): """ Called by the libres context when the session provider is being discarded (only in testing). This makes sure that replacing the session provider on the context doesn't leave behind any idle connections. """ self.session().close() self.engine.raw_connection().invalidate() self.engine.dispose()
[docs] def get_postgres_version(self, dsn): """ Returns the postgres version in a tuple with the first value being the major version, the second being the minor version. Uses it's own connection to be independent from any session. """ assert 'postgres' in dsn, "Not a postgres database" engine = create_engine(dsn) version = engine.execute('select version()').fetchone()[0] engine.dispose() version = re.findall('PostgreSQL (.*?) on', version)[0] return [int(fragment) for fragment in version.split('.')][:2]
def assert_valid_postgres_version(self, dsn): major, minor = self.get_postgres_version(dsn) assert (major >= 9 and minor >= 1) or (major >= 10), \ "PostgreSQL 9.1+ is required. Your version is {}.{}".format( major, minor) return dsn