Python replica_set_connection.ReplicaSetConnection类代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

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

本文整理汇总了Python中pymongo.replica_set_connection.ReplicaSetConnection类的典型用法代码示例。如果您正苦于以下问题：Python ReplicaSetConnection类的具体用法？Python ReplicaSetConnection怎么用？Python ReplicaSetConnection使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

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

示例1: test_replica_set_connection

    def test_replica_set_connection(self):
        c = ReplicaSetConnection(pair, replicaSet=self.name)
        self.assertTrue(c.auto_start_request)
        self.assertFalse(c.slave_okay)
        self.assertFalse(c.safe)
        self.assertEqual({}, c.get_lasterror_options())

        # ReplicaSetConnection's writes are unacknowledged by default
        doc = {"_id": ObjectId()}
        coll = c.pymongo_test.write_concern_test
        coll.really_drop()
        coll.insert(doc)
        coll.insert(doc)

        c = ReplicaSetConnection("mongodb://%s:%s/?replicaSet=%s&safe=true" % (
            host, port, self.name))

        self.assertTrue(c.safe)

        # ReplicaSetConnection's network_timeout argument is translated into
        # socketTimeoutMS
        self.assertEqual(123, ReplicaSetConnection(
            pair, replicaSet=self.name, network_timeout=123
        )._MongoReplicaSetClient__net_timeout)

        for network_timeout in 'foo', 0, -1:
            self.assertRaises(ConfigurationError,
                ReplicaSetConnection, pair, replicaSet=self.name,
                network_timeout=network_timeout)

开发者ID:conversocial，项目名称:mongo-python-driver，代码行数:29，代码来源:test_legacy_connections.py

示例2: test_ipv6

    def test_ipv6(self):
        try:
            connection = ReplicaSetConnection("[::1]:%d" % (port,),
                                              replicaSet=self.name)
        except:
            # Either mongod was started without --ipv6
            # or the OS doesn't support it (or both).
            raise SkipTest()

        # Try a few simple things
        connection = ReplicaSetConnection("mongodb://[::1]:%d" % (port,),
                                          replicaSet=self.name)
        connection = ReplicaSetConnection("mongodb://[::1]:%d/?safe=true;"
                                          "replicaSet=%s" % (port, self.name))
        connection = ReplicaSetConnection("[::1]:%d,localhost:"
                                          "%d" % (port, port),
                                          replicaSet=self.name)
        connection = ReplicaSetConnection("localhost:%d,[::1]:"
                                          "%d" % (port, port),
                                          replicaSet=self.name)
        connection.pymongo_test.test.save({"dummy": u"object"})
        connection.pymongo_test_bernie.test.save({"dummy": u"object"})

        dbs = connection.database_names()
        self.assertTrue("pymongo_test" in dbs)
        self.assertTrue("pymongo_test_bernie" in dbs)
        connection.close()

开发者ID:Basis，项目名称:mongo-python-driver，代码行数:27，代码来源:test_replica_set_connection.py

示例3: test_replica_set_connection

    def test_replica_set_connection(self):
        c = ReplicaSetConnection(pair, replicaSet=self.name)

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertTrue(c.auto_start_request)
            self.assertEqual(None, c.max_pool_size)
            self.assertFalse(c.slave_okay)
            self.assertFalse(c.safe)
            self.assertEqual({}, c.get_lasterror_options())

            # ReplicaSetConnection's writes are unacknowledged by default
            doc = {"_id": ObjectId()}
            coll = c.pymongo_test.write_concern_test
            coll.drop()
            coll.insert(doc)
            coll.insert(doc)

            c = ReplicaSetConnection("mongodb://%s:%s/?replicaSet=%s&safe=true" % (
                host, port, self.name))

            self.assertTrue(c.safe)
        finally:
            ctx.exit()

        # To preserve legacy ReplicaSetConnection's behavior, max_size should
        # be None. Pool should handle this without error.
        pool = get_pool(c)
        self.assertEqual(None, pool.max_size)
        c.end_request()

        # ReplicaSetConnection's network_timeout argument is translated into
        # socketTimeoutMS
        self.assertEqual(123, ReplicaSetConnection(
            pair, replicaSet=self.name, network_timeout=123
        )._MongoReplicaSetClient__net_timeout)

        for network_timeout in 'foo', 0, -1:
            self.assertRaises(
                ConfigurationError,
                ReplicaSetConnection, pair, replicaSet=self.name,
                network_timeout=network_timeout)

开发者ID:hedgepigdaniel，项目名称:mongo-python-driver，代码行数:43，代码来源:test_legacy_connections.py

示例4: test_properties

    def test_properties(self):
        c = ReplicaSetConnection(pair, replicaSet=self.name)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.read_preference, ReadPreference.PRIMARY)
        self.assertEqual(c.max_pool_size, 10)
        self.assertEqual(c.document_class, dict)
        self.assertEqual(c.tz_aware, False)
        self.assertEqual(c.slave_okay, False)
        self.assertEqual(c.safe, False)
        c.close()

        c = ReplicaSetConnection(pair, replicaSet=self.name, max_pool_size=25,
                                 document_class=SON, tz_aware=True,
                                 slaveOk=False, safe=True,
                                 read_preference=ReadPreference.SECONDARY)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.read_preference, ReadPreference.SECONDARY)
        self.assertEqual(c.max_pool_size, 25)
        self.assertEqual(c.document_class, SON)
        self.assertEqual(c.tz_aware, True)
        self.assertEqual(c.slave_okay, False)
        self.assertEqual(c.safe, True)

        if version.at_least(c, (1, 7, 4)):
            self.assertEqual(c.max_bson_size, 16777216)
        else:
            self.assertEqual(c.max_bson_size, 4194304)
        c.close()

开发者ID:samuelclay，项目名称:mongo-python-driver，代码行数:34，代码来源:test_replica_set_connection.py

示例5: test_ipv6

    def test_ipv6(self):
        try:
            connection = ReplicaSetConnection("[::1]:27017", replicaSet=self.name)
        except:
            # Either mongod was started without --ipv6
            # or the OS doesn't support it (or both).
            raise SkipTest()

        # Try a few simple things
        connection = ReplicaSetConnection("mongodb://[::1]:27017",
                                          replicaSet=self.name)
        uri = "mongodb://[::1]:27017/?slaveOk=true;replicaSet=" + self.name
        connection = ReplicaSetConnection(uri)
        connection = ReplicaSetConnection("[::1]:27017,localhost:27017",
                                          replicaSet=self.name)
        connection = ReplicaSetConnection("localhost:27017,[::1]:27017",
                                          replicaSet=self.name)
        connection.pymongo_test.test.save({"dummy": u"object"})
        connection.pymongo_test_bernie.test.save({"dummy": u"object"})

        dbs = connection.database_names()
        self.assert_("pymongo_test" in dbs)
        self.assert_("pymongo_test_bernie" in dbs)
        connection.close()

开发者ID:samuelclay，项目名称:mongo-python-driver，代码行数:24，代码来源:test_replica_set_connection.py

示例6: test_properties

    def test_properties(self):
        c = ReplicaSetConnection(pair, replicaSet=self.name)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.max_pool_size, 10)
        self.assertEqual(c.document_class, dict)
        self.assertEqual(c.tz_aware, False)

        # Make sure RSC's properties are copied to Database and Collection
        for obj in c, c.pymongo_test, c.pymongo_test.test:
            self.assertEqual(obj.read_preference, ReadPreference.PRIMARY)
            self.assertEqual(obj.tag_sets, [{}])
            self.assertEqual(obj.secondary_acceptable_latency_ms, 15)
            self.assertEqual(obj.slave_okay, False)
            self.assertEqual(obj.safe, False)

        cursor = c.pymongo_test.test.find()
        self.assertEqual(
            ReadPreference.PRIMARY, cursor._Cursor__read_preference)
        self.assertEqual([{}], cursor._Cursor__tag_sets)
        self.assertEqual(15, cursor._Cursor__secondary_acceptable_latency_ms)
        self.assertEqual(False, cursor._Cursor__slave_okay)
        c.close()

        tag_sets = [{'dc': 'la', 'rack': '2'}, {'foo': 'bar'}]
        c = ReplicaSetConnection(pair, replicaSet=self.name, max_pool_size=25,
                                 document_class=SON, tz_aware=True,
                                 slaveOk=False, safe=True,
                                 read_preference=ReadPreference.SECONDARY,
                                 tag_sets=copy.deepcopy(tag_sets),
                                 secondary_acceptable_latency_ms=77)
        c.admin.command('ping')
        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.hosts, self.hosts)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.max_pool_size, 25)
        self.assertEqual(c.document_class, SON)
        self.assertEqual(c.tz_aware, True)

        for obj in c, c.pymongo_test, c.pymongo_test.test:
            self.assertEqual(obj.read_preference, ReadPreference.SECONDARY)
            self.assertEqual(obj.tag_sets, tag_sets)
            self.assertEqual(obj.secondary_acceptable_latency_ms, 77)
            self.assertEqual(obj.slave_okay, False)
            self.assertEqual(obj.safe, True)

        cursor = c.pymongo_test.test.find()
        self.assertEqual(
            ReadPreference.SECONDARY, cursor._Cursor__read_preference)
        self.assertEqual(tag_sets, cursor._Cursor__tag_sets)
        self.assertEqual(77, cursor._Cursor__secondary_acceptable_latency_ms)
        self.assertEqual(False, cursor._Cursor__slave_okay)

        cursor = c.pymongo_test.test.find(
            read_preference=ReadPreference.NEAREST,
            tag_sets=[{'dc':'ny'}, {}],
            secondary_acceptable_latency_ms=123)

        self.assertEqual(
            ReadPreference.NEAREST, cursor._Cursor__read_preference)
        self.assertEqual([{'dc':'ny'}, {}], cursor._Cursor__tag_sets)
        self.assertEqual(123, cursor._Cursor__secondary_acceptable_latency_ms)
        self.assertEqual(False, cursor._Cursor__slave_okay)

        if version.at_least(c, (1, 7, 4)):
            self.assertEqual(c.max_bson_size, 16777216)
        else:
            self.assertEqual(c.max_bson_size, 4194304)
        c.close()

开发者ID:SiroDiazPalazon，项目名称:mongo-python-driver，代码行数:71，代码来源:test_replica_set_connection.py

示例7: init

 def __init__(self, *args, **kwargs):
     # Specifying that it should run both the inits
     MongoKitConnection.__init__(self, *args, **kwargs)
     PymongoReplicaSetConnection.__init__(self, *args, **kwargs)

开发者ID:Ink，项目名称:mongokit，代码行数:4，代码来源:replica_set_connection.py

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

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python topology.Topology类代码示例发布时间：2022-05-27

Python read_preferences.MovingAverage类代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13807|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10200|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4091|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4043|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3845|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3514|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3031|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2655|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2650|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2302|2022-01-22

客服电话

电子邮件

Python replica_set_connection.ReplicaSetConnection类代码示例

示例1: test_replica_set_connection

示例2: test_ipv6

示例3: test_replica_set_connection

示例4: test_properties

示例5: test_ipv6

示例6: test_properties

示例7: init

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053

客服电话

电子邮件

Python replica_set_connection.ReplicaSetConnection类代码示例

示例1: test_replica_set_connection

示例2: test_ipv6

示例3: test_replica_set_connection

示例4: test_properties

示例5: test_ipv6

示例6: test_properties

示例7: __init__

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053

示例7: init