How to use the pymongo.ReadPreference function in pymongo

sys.stdout.write(first_line)

        logger = logging.Logger("", level=logging.DEBUG)
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(logging.Formatter(fmt="%(message)s"))
        logger.addHandler(handler)

        self._stdout_pipe = pipe.LoggerPipe(logger, logging.INFO, self.proc.stdout)
        self._stdout_pipe.wait_until_started()

        if not self.did_mongod_start(self.port):
            raise Exception("Failed to start mongod")

        if self.slave:
            local = MongoClient(port=self.port,
                read_preference=ReadPreference.SECONDARY_PREFERRED).local
            synced = False
            while not synced:
                synced = True
                for source in local.sources.find(fields=["syncedTo"]):
                    synced = synced and "syncedTo" in source and source["syncedTo"]

def test_parallel_scan(self):
        if not (yield at_least(self.cx, (2, 5, 5))):
            raise SkipTest("Requires MongoDB >= 2.5.5")

        yield skip_if_mongos(self.cx)

        collection = self.collection

        # Enough documents that each cursor requires multiple batches.
        yield collection.remove()
        yield collection.insert(({'_id': i} for i in range(8000)), w=test.env.w)
        if test.env.is_replica_set:
            client = self.motor_rsc()

            # Test that getMore messages are sent to the right server.
            client.read_preference = ReadPreference.SECONDARY
            collection = client.motor_test.test_collection

        docs = []

        @gen.coroutine
        def f(cursor):
            self.assertTrue(isinstance(cursor,
                                       motor.motor_tornado.MotorCommandCursor))

            while (yield cursor.fetch_next):
                docs.append(cursor.next_object())

        cursors = yield collection.parallel_scan(3)
        yield [f(cursor) for cursor in cursors]
        self.assertEqual(len(docs), (yield collection.count()))

def setUp(self):
        self.master = MongoClient(host, port)

        self.slaves = []
        try:
            self.slaves.append(MongoClient(
                host2, port2, read_preference=ReadPreference.SECONDARY))
        except ConnectionFailure:
            pass

        try:
            self.slaves.append(MongoClient(
                host3, port3, read_preference=ReadPreference.SECONDARY))
        except ConnectionFailure:
            pass

        if not self.slaves:
            raise SkipTest("Not connected to master-slave set")

        self.client = MasterSlaveConnection(self.master, self.slaves)
        self.db = self.client.pymongo_test

c = MockClient(
                standalones=[],
                members=['a:1', 'b:2'],
                mongoses=[],
                host='a:1',
                replicaSet='rs')
            self.addCleanup(c.close)

            wait_until(lambda: len(c.nodes) == 2, 'discover both nodes')

            # b now raises socket.error.
            c.mock_down_hosts.append('b:2')
            self.assertRaises(
                ConnectionFailure,
                c.db.collection.with_options(
                    read_preference=ReadPreference.SECONDARY).find_one)

            self.assertEqual(1, len(c.nodes))

:class:`~datetime.datetime` instances returned as values
            in a document by this :class:`MasterSlaveConnection` will be timezone
            aware (otherwise they will be naive)
        """
        if not isinstance(master, MongoClient):
            raise TypeError("master must be a MongoClient instance")
        if not isinstance(slaves, list) or len(slaves) == 0:
            raise TypeError("slaves must be a list of length >= 1")

        for slave in slaves:
            if not isinstance(slave, MongoClient):
                raise TypeError("slave %r is not an instance of MongoClient" %
                                slave)

        super(MasterSlaveConnection,
              self).__init__(read_preference=ReadPreference.SECONDARY,
                             safe=master.safe,
                             **master.write_concern)

        self.__master = master
        self.__slaves = slaves
        self.__document_class = document_class
        self.__tz_aware = tz_aware
        self.__request_counter = thread_util.Counter(master.use_greenlets)

def extract(self):
        for a in (
            Vendor.objects.filter(read_preference=ReadPreference.SECONDARY_PREFERRED)
            .all()
            .order_by("id")
        ):
            yield (a.bi_id, a.id, a.name)

def extract(self):
        pipeline = [
            {"$group": {"_id": "$object", "last_ts": {"$max": "$ts"}}},
            {"$sort": {"_id": 1}},
        ]
        if len(self.sync_ids) < 20000:
            # @todo Very large list slowest encode, need research
            pipeline.insert(0, {"$match": {"object": {"$in": self.sync_ids}}})
        value = (
            get_db()["noc.gridvcs.config.files"]
            .with_options(read_preference=ReadPreference.SECONDARY_PREFERRED)
            .aggregate(pipeline)
        )
        for v in value:
            if not v["_id"]:
                continue
            yield v["_id"], v["last_ts"]

def __iter__(self):
        for p in self.mos_pools:
            r = (
                get_db()[self.coll_name % p.name]
                .with_options(read_preference=ReadPreference.SECONDARY_PREFERRED)
                .aggregate(self.pipelines.get(p.name, self.pipeline()))
            )
            for x in r:
                # @todo Append info for MO
                yield x

from werkzeug.routing import BaseConverter
import pymongo
import warnings

from flask_pymongo.wrappers import Connection
from flask_pymongo.wrappers import ReplicaSetConnection


PRIMARY = pymongo.ReadPreference.PRIMARY
"""Send all queries to the replica set primary, and fail if none exists."""

SECONDARY = pymongo.ReadPreference.SECONDARY
"""Distribute queries among replica set secondaries unless none exist or
are up, in which case send queries to the primary."""

SECONDARY_ONLY = pymongo.ReadPreference.SECONDARY_ONLY
"""Distribute queries among replica set secondaries, and fail if none
exist."""

DESCENDING = pymongo.DESCENDING
"""Descending sort order."""

ASCENDING = pymongo.ASCENDING
"""Ascending sort order."""

READ_PREFERENCE_MAP = {
    # this handles defaulting to PRIMARY for us
    None: PRIMARY,

    # alias the string names to the correct constants
    'PRIMARY': PRIMARY,
    'SECONDARY': SECONDARY,

def extract(self):
        for a in (
            SubscriberProfile.objects.filter(read_preference=ReadPreference.SECONDARY_PREFERRED)
            .all()
            .order_by("id")
        ):
            yield (a.bi_id, a.id, a.name, a.description, a.glyph)