def get_session(self):
     if not self.sessionmaker:
         engine = self.get_db_engine()
         self.sessionmaker = scoped_session(sessionmaker(bind=engine))
     session = self.sessionmaker()
     session.rollback()
     return session

    def setUp(self):
        """Creates the Flask application and the APIManager."""
        # create the Flask application
        app = Flask(__name__)
        app.config['DEBUG'] = True
        app.config['TESTING'] = True
        app.config['SECRET_KEY'] = "testing..."
        del app.logger.handlers[0]
        # sqlalchemy flask
        self.base = declarative_base()
        self.engine = create_engine('sqlite://')
        self.session = scoped_session(sessionmaker(autocommit=False,
                                                   autoflush=False,
                                                   bind=self.engine))
        self.app = app
        self._ctx = self.app.test_request_context()
        self._ctx.push()
        self.client = self.app.test_client()

        # Add an error handler that returns straight LeverException
        # recommendations
        @self.app.errorhandler(LeverException)
        def handler(exc):
            tb = exc.extra.pop('tb', None)
            self.app.logger.debug("Lever Exception Thrown!\n"
                                  "Extra: {0}\nEnd User: {1}"
                                  .format(exc.extra, exc.end_user),
                                    exc_info=True)
            if tb:
                self.app.logger.debug("Traceback from lever exception:"
                                      "\n{0}".format(tb.replace('\\n', '\n')))
            return jsonify(**exc.end_user), exc.code

        return app

 def session(self, dbname):
     # set up a session for each db; this uses scoped_session (based on the
     # thread ID) to ensure only one session per thread
     if dbname not in self._sessions:
         Session = orm.sessionmaker(bind=self.engine(dbname))
         self._sessions[dbname] = scoping.scoped_session(Session)
     return self._sessions[dbname]

    def __init__(self,keep_session=False):
        #cache
        self.dash_engines = []

        #check db
        if len(settings.SQL_BACKENDS)==0:
            raise Exception('Settings SQL_BACKENDS need one db at least!')

        #初始化数据库连接
        #初始化session marker
        for backend in settings.SQL_BACKENDS:
            engine = create_engine(
                backend['db_url'],
                pool_size=backend['db_pool_size'],
                echo=settings.SQL_DEBUG,
                encoding=backend['db_charset'],
                pool_recycle=backend['db_pool_recycle']
            )
            self.dash_engines.append(engine)

        #初始化shardstat
        idx=0
        s=scoped_session(sessionmaker(bind=self.dash_engines[0]))()
        for db_setting in settings.SQL_BACKENDS:
            ss=ShardStat(shard_name=db_setting['db_name'],
                shard_id=idx)
            s.add(ss)
            idx+=1
            try:
                s.commit()
            except:
                s.rollback()

        s.close()

 def _create_sessions(self):
     ''' Creates database sessions '''
     gi_tax_engine = create_engine(self.gi_taxonomy_url, echo=False,
         convert_unicode=True, encoding='utf-8',
         pool_recycle=3600)
     gi_tax_session = scoped_session(sessionmaker(
         bind=gi_tax_engine, autocommit=False, autoflush=False))
     self.gi_tax_session = gi_tax_session

def session(app):
    sess = None
    try:
        sess = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=app.engine))
        yield sess
    finally:
        if sess:
            sess.close()

 def __init__(self, db_connection, library_path, login, slot_id):
     self.dry_run = False
     self.db_engine = sqlalchemy.create_engine(db_connection)
     self._session_creator = scoping.scoped_session(orm.sessionmaker(bind=self.db_engine, autocommit=True))
     self.crypto_plugin = p11_crypto.P11CryptoPlugin(CONF)
     self.plugin_name = utils.generate_fullname_for(self.crypto_plugin)
     self.pkcs11 = self.crypto_plugin.pkcs11
     self.session = self.pkcs11.get_session()

 def start(self):
     db_dir_path = self.resource_manager.get_fs_resource_path("store")
     self.db_string = "sqlite:///%s/app.db" % db_dir_path
     self.engine = create_engine(self.db_string)
     self.session = scoped_session(sessionmaker(bind=self.engine,
                                                autocommit=False,
                                                autoflush=True))
     Base.metadata.create_all(self.engine)

    def _create_sessions(self):
        ''' Creates database sessions '''
        unity_engine = create_engine (self.unity_db_url, echo=False, 
                                convert_unicode=True, encoding='utf-8',
                                pool_recycle=3600)
        unity_session = scoped_session(sessionmaker(
                        bind=unity_engine, autocommit=False, autoflush=False))
        
        self.unity_session = unity_session
        
        ncbitax_engine = create_engine (self.ncbitax_db_url, echo=False, 
                                convert_unicode=True, encoding='utf-8',
                                pool_recycle=3600)
        
        ncbitax_session = scoped_session(sessionmaker(
                        bind=ncbitax_engine, autocommit=False, autoflush=False))

        self.ncbitax_session = ncbitax_session

        def SelectAll(self,pModel):

                try:
                        Session = scoped_session(sessionmaker(bind=connect.ConnectorMySql()))
                        ses = Session()
                        lst =  ses.query(pModel).all()
                        return lst

                except Exception as e:
                        raise Exception('Type Error In SelectAll :'+ str(e))

    def __init__(self, connection=None):
        self.connection = connection or cfg.db.connection

        self.engine = sqlalchemy.create_engine(self.connection)
        self.DBSession = scoping.scoped_session(
            orm.sessionmaker(
                bind=self.engine,
                autocommit=True
            )
        )

 def dispatch(self):
     """Add the database session to the request's scope."""
     try:
         self.db_session = scoped_session(sessionmaker(bind=self.engine))
         ret = super(BaseHandler, self).dispatch()
         self.db_session.commit()
         return ret
     except:
         self.db_session.rollback()
         raise

def init_session(database_filename):
    engine = create_engine('sqlite:///{0}'.format(database_filename), echo=False)
    DBSession = scoped_session(
        sessionmaker(
            autoflush=True,
            autocommit=False,
            bind=engine
        )
    )
    return DBSession

    def init(self):
        '''Initialize(reset) mocked in-memory copy.
        '''
        self.url = 'sqlite://'
        self.mock_engine = sqlalchemy.create_engine(self.url)

        self.MBase = declarative_base()
        self.MBase.metadata = self.TBase.metadata
        self.MBase.metadata.bind = self.mock_engine
        self.MBase.metadata.create_all()
        self._Session = scoped_session(sessionmaker(bind=self.mock_engine))

 def __init__(self):
     self.nv_log_handler = nv_logger(self.__class__.__name__).get_logger()
     self.db_engine = create_engine(NVDB_SQLALCHEMY_DB,
                                 connect_args={'check_same_thread': False},
                                 poolclass=StaticPool, echo=False)
     session_maker = sessionmaker(bind=self.db_engine)
     self.Session = scoped_session(session_maker)
     self.db_session = None
     db_base.metadata.create_all(self.db_engine)
     self.nv_midbox_db_entry = None
     self.nv_webserver = None
     self.nv_log_handler.debug("Tables created in nvdb")

        def InsertWithOutGetID(self,pmodel):
            try:
                Session = scoped_session(sessionmaker(bind=connect.ConnectorMySql()))
                Base.query = Session.query_property()
                ses = Session()
                ses.add(pmodel)
                ses.commit()
                ses.close()
                return 

            except Exception as e:

                raise Exception('Type Error In Insert :'+ str(e))

    def _init_db(self):
        print '>>>', 'init_db'
        db_config_file = os.path.abspath('config/database.py')
        # execfile(db_config_file)
        db_info = {'username':'root', 
           'password':'123456',
           'host':'127.0.0.1',
           'db':'storm'}

        engine = create_engine('mysql://%s:%[email protected]%s/%s?charset=utf8' % (db_info['username'], db_info['password'], db_info['host'], db_info['db']),
                                echo=False)
        # Session = sessionmaker(bind=engine)
        self.db = scoped_session(sessionmaker(bind=engine))

        def __init__(self):
            """
            dialect+driver://username:[email protected]:port/database
            """

            self.conf = SysUtil().get_sys_conf()
            db_username = self.conf["db_username"]
            db_password = self.conf["db_password"]
            db_url = self.conf["db_url"]
            db_name = self.conf["db_name"]

            self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
            self._scoped_session = scoped_session(sessionmaker(bind=self.engine, expire_on_commit=True))
            self.session = None

def open_db_connection(dsn):
    """
        Sets up and configures the SQLAlchemy session using the provided L{dsn}.

        :param dsn: The DSN to connect too
        :type dsn: str
    """
    assert isinstance(dsn, str)
    assert dsn

    engine = create_engine(dsn, echo=False, pool_recycle=3600)

    global Session
    Session = scoped_session(sessionmaker(bind=engine, autocommit=True))

def initialize(settings, **kwargs):
    create_db = kwargs.get('create_db', settings['app.create_db'])
    engine = engine_from_config(
        settings, prefix='sa.', echo=settings['app.echo']
    )
    maker = scoped_session(sessionmaker(
        extension=ZopeTransactionExtension()
    ))
    maker.configure(bind=engine)
    db.register_maker(maker)
    Base.metadata.bind = engine

    if create_db:
        logger.info(u"Flag 'app.create_db' is ON. Creating tables ...")
        Base.metadata.create_all(engine)