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

Python engines.reconnecting_engine函数代码示例

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

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



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

示例1: test_pre_ping_db_is_restarted

    def test_pre_ping_db_is_restarted(self):
        engine = engines.reconnecting_engine(
            options={"pool_pre_ping": True}
        )

        conn = engine.connect()
        eq_(conn.execute(select([1])).scalar(), 1)
        stale_connection = conn.connection.connection
        conn.close()

        engine.test_shutdown()
        engine.test_restart()

        conn = engine.connect()
        eq_(conn.execute(select([1])).scalar(), 1)
        conn.close()

        def exercise_stale_connection():
            curs = stale_connection.cursor()
            curs.execute("select 1")

        assert_raises(
            engine.dialect.dbapi.Error,
            exercise_stale_connection
        )
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:25,代码来源:test_reconnect.py


示例2: test_basic

    def test_basic(self):
        for threadlocal in False, True:
            engine = engines.reconnecting_engine(
                options={'pool_threadlocal': threadlocal})

            conn = engine.contextual_connect()
            eq_(conn.execute(select([1])).scalar(), 1)
            conn.close()

            # set the pool recycle down to 1.
            # we aren't doing this inline with the
            # engine create since cx_oracle takes way
            # too long to create the 1st connection and don't
            # want to build a huge delay into this test.

            engine.pool._recycle = 1

            # kill the DB connection
            engine.test_shutdown()

            # wait until past the recycle period
            time.sleep(2)

            # can connect, no exception
            conn = engine.contextual_connect()
            eq_(conn.execute(select([1])).scalar(), 1)
            conn.close()
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:27,代码来源:test_reconnect.py


示例3: setup

 def setup(self):
     self.engine = engines.reconnecting_engine()
     self.meta = MetaData(self.engine)
     table = Table('sometable', self.meta,
                   Column('id', Integer, primary_key=True),
                   Column('name', String(50)))
     self.meta.create_all()
     table.insert().execute(
         [{'id': i, 'name': 'row %d' % i} for i in range(1, 100)]
     )
开发者ID:SvenDowideit,项目名称:clearlinux,代码行数:10,代码来源:test_reconnect.py


示例4: test_pre_ping_db_stays_shutdown

    def test_pre_ping_db_stays_shutdown(self):
        engine = engines.reconnecting_engine(options={"pool_pre_ping": True})

        conn = engine.connect()
        eq_(conn.execute(select([1])).scalar(), 1)
        conn.close()

        engine.test_shutdown(stop=True)

        assert_raises(exc.DBAPIError, engine.connect)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:10,代码来源:test_reconnect.py


示例5: test_null_pool

 def test_null_pool(self):
     engine = engines.reconnecting_engine(options=dict(poolclass=pool.NullPool))
     conn = engine.connect()
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.closed
     engine.test_shutdown()
     _assert_invalidated(conn.execute, select([1]))
     assert not conn.closed
     assert conn.invalidated
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.invalidated
开发者ID:t3573393,项目名称:sqlalchemy,代码行数:11,代码来源:test_reconnect.py


示例6: test_null_pool

 def test_null_pool(self):
     engine = \
         engines.reconnecting_engine(options=dict(poolclass=pool.NullPool))
     conn = engine.connect()
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.closed
     engine.test_shutdown()
     try:
         conn.execute(select([1]))
         assert False
     except tsa.exc.DBAPIError, e:
         if not e.connection_invalidated:
             raise
开发者ID:pshken,项目名称:sqlalchemy,代码行数:13,代码来源:test_reconnect.py


示例7: setup

 def setup(self):
     self.engine = engines.reconnecting_engine()
     self.meta = MetaData(self.engine)
     table = Table(
         "sometable",
         self.meta,
         Column("id", Integer, primary_key=True),
         Column("name", String(50)),
     )
     self.meta.create_all()
     table.insert().execute(
         [{"id": i, "name": "row %d" % i} for i in range(1, 100)]
     )
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:13,代码来源:test_reconnect.py


示例8: setup

 def setup(self):
     self.engine = engines.reconnecting_engine()
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:2,代码来源:test_reconnect.py


示例9: setup

 def setup(self):
     global engine
     engine = engines.reconnecting_engine()
开发者ID:pshken,项目名称:sqlalchemy,代码行数:3,代码来源:test_reconnect.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python engines.testing_engine函数代码示例发布时间:2022-05-27
下一篇:
Python assertions.eq_函数代码示例发布时间: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