• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python scoping.scoped_session函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中sqlalchemy.orm.scoping.scoped_session函数的典型用法代码示例。如果您正苦于以下问题:Python scoped_session函数的具体用法?Python scoped_session怎么用?Python scoped_session使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了scoped_session函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: get_session

 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
开发者ID:7scientists,项目名称:rouster,代码行数:7,代码来源:environment.py


示例2: setUp

    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
开发者ID:Crowdlink,项目名称:lever,代码行数:34,代码来源:model_helpers.py


示例3: session

 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]
开发者ID:b10n1k,项目名称:build-relengapi,代码行数:7,代码来源:db.py


示例4: __init__

    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()
开发者ID:niyoufa,项目名称:ods,代码行数:34,代码来源:models.py


示例5: _create_sessions

 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
开发者ID:abulovic,项目名称:metagenomix,代码行数:8,代码来源:access.py


示例6: 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()
开发者ID:rizac,项目名称:gfz-reportgen,代码行数:8,代码来源:models.py


示例7: __init__

 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()
开发者ID:john5223,项目名称:barbican,代码行数:8,代码来源:pkcs11_migrate_kek_signatures.py


示例8: start

 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)
开发者ID:swstack,项目名称:stephenstack.com.v2,代码行数:8,代码来源:database.py


示例9: _create_sessions

    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
开发者ID:abulovic,项目名称:binner,代码行数:18,代码来源:access.py


示例10: SelectAll

        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))
开发者ID:RaminFP,项目名称:SQLAlchemy_RestfulAPI,代码行数:10,代码来源:repository.py


示例11: __init__

    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
            )
        )
开发者ID:chadlung,项目名称:battlement,代码行数:10,代码来源:__init__.py


示例12: dispatch

 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
开发者ID:LyleScott,项目名称:interview-project-contact-manager,代码行数:10,代码来源:contact_manager.py


示例13: init_session

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
开发者ID:llazzaro,项目名称:lcubo_helpers,代码行数:10,代码来源:__init__.py


示例14: init

    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))
开发者ID:Snoin,项目名称:seektam-web,代码行数:11,代码来源:mock.py


示例15: __init__

 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")
开发者ID:sugchand,项目名称:neoview-middle-box,代码行数:12,代码来源:nvdb_manager.py


示例16: InsertWithOutGetID

        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))
开发者ID:RaminFP,项目名称:SQLAlchemy_RestfulAPI,代码行数:13,代码来源:repository.py


示例17: _init_db

    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))
开发者ID:jungledrum,项目名称:noodle,代码行数:13,代码来源:application.py


示例18: __init__

        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
开发者ID:MobileCloudNetworking,项目名称:imsaas,代码行数:14,代码来源:DatabaseManager.py


示例19: open_db_connection

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))
开发者ID:sgiroux,项目名称:trex,代码行数:14,代码来源:__init__.py


示例20: initialize

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)
开发者ID:jnosal,项目名称:eventlog,代码行数:15,代码来源:__init__.py



注:本文中的sqlalchemy.orm.scoping.scoped_session函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python session.make_transient函数代码示例发布时间:2022-05-27
下一篇:
Python query.Query类代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap