python类version()的实例源码

def check_compatibility(module, client):
    """Check the compatibility between the driver and the database.

       See: https://docs.mongodb.com/ecosystem/drivers/driver-compatibility-reference/#python-driver-compatibility

    Args:
        module: Ansible module.
        client (cursor): Mongodb cursor on admin database.
    """
    loose_srv_version = LooseVersion(client.server_info()['version'])
    loose_driver_version = LooseVersion(PyMongoVersion)

    if loose_srv_version >= LooseVersion('3.2') and loose_driver_version <= LooseVersion('3.2'):
        module.fail_json(msg=' (Note: you must use pymongo 3.2+ with MongoDB >= 3.2)')

    elif loose_srv_version >= LooseVersion('3.0') and loose_driver_version <= LooseVersion('2.8'):
        module.fail_json(msg=' (Note: you must use pymongo 2.8+ with MongoDB 3.0)')

    elif loose_srv_version >= LooseVersion('2.6') and loose_driver_version <= LooseVersion('2.7'):
        module.fail_json(msg=' (Note: you must use pymongo 2.7+ with MongoDB 2.6)')

    elif LooseVersion(PyMongoVersion) <= LooseVersion('2.5'):
        module.fail_json(msg=' (Note: you must be on mongodb 2.4+ and pymongo 2.5+ to use the roles param)')

def mongo_connect(host=None, port=None,ssl=False, user=None,passwd=None,replica=None):
    try:
        # ssl connection for pymongo > 2.1
        if pymongo.version >= "2.1":
            if replica is None:
                con = pymongo.Connection(host, port, read_preference=pymongo.ReadPreference.SECONDARY, ssl=ssl, network_timeout=10)
            else:
                con = pymongo.Connection(host, port, read_preference=pymongo.ReadPreference.SECONDARY, ssl=ssl, replicaSet=replica, network_timeout=10)
        else:
            if replica is None:
                con = pymongo.Connection(host, port, slave_okay=True, network_timeout=10)
            else:
                con = pymongo.Connection(host, port, slave_okay=True, replicaSet=replica, network_timeout=10)

        if user and passwd:
            db = con["admin"]
            if not db.authenticate(user, passwd): sys.exit("Username/Password incorrect")
    except Exception, e:
        if isinstance(e,pymongo.errors.AutoReconnect) and str(e).find(" is an arbiter") != -1:
            # We got a pymongo AutoReconnect exception that tells us we connected to an Arbiter Server
            # This means: Arbiter is reachable and can answer requests/votes - this is all we need to know from an arbiter
            print "OK - State: 7 (Arbiter)"
            sys.exit(0)
        return exit_with_general_critical(e),None
    return 0,con

def __init__(self, attr_name, has_write_concern, doc=None):
        """A descriptor that wraps a PyMongo method, such as insert or remove,
        and returns an asynchronous version of the method, which accepts a
        callback or returns a Future.

        :Parameters:
         - `attr_name`: The name of the attribute on the PyMongo class, if
           different from attribute on the Motor class
         - `has_write_concern`: Whether the method accepts getLastError options
        """
        super(Async, self).__init__(doc)
        self.attr_name = attr_name
        self.has_write_concern = has_write_concern

def parallel_scan(self, num_cursors, **kwargs):
        """Scan this entire collection in parallel.

        Returns a list of up to ``num_cursors`` cursors that can be iterated
        concurrently. As long as the collection is not modified during
        scanning, each document appears once in one of the cursors' result
        sets.

        For example, to process each document in a collection using some
        function ``process_document()``::

            @gen.coroutine
            def process_cursor(cursor):
                while (yield cursor.fetch_next):
                    process_document(document)

            # Get up to 4 cursors.
            cursors = yield collection.parallel_scan(4)
            yield [process_cursor(cursor) for cursor in cursors]

            # All documents have now been processed.

        If ``process_document()`` is a coroutine, do
        ``yield process_document(document)``.

        With :class:`MotorReplicaSetClient`, pass `read_preference` of
        :attr:`~pymongo.read_preference.ReadPreference.SECONDARY_PREFERRED`
        to scan a secondary.

        :Parameters:
          - `num_cursors`: the number of cursors to return

        .. note:: Requires server version **>= 2.5.5**.
        """
        command_cursors = yield self.__parallel_scan(num_cursors, **kwargs)
        motor_command_cursors = [
            MotorCommandCursor(cursor, self)
            for cursor in command_cursors]

        raise gen.Return(motor_command_cursors)

def __init__(self, attr_name, has_write_concern, doc=None):
        """A descriptor that wraps a PyMongo method, such as insert or remove,
        and returns an asynchronous version of the method, which accepts a
        callback or returns a Future.

        :Parameters:
         - `attr_name`: The name of the attribute on the PyMongo class, if
           different from attribute on the Motor class
         - `has_write_concern`: Whether the method accepts getLastError options
        """
        super(Async, self).__init__(doc)
        self.attr_name = attr_name
        self.has_write_concern = has_write_concern

def parallel_scan(self, num_cursors, **kwargs):
        """Scan this entire collection in parallel.

        Returns a list of up to ``num_cursors`` cursors that can be iterated
        concurrently. As long as the collection is not modified during
        scanning, each document appears once in one of the cursors' result
        sets.

        For example, to process each document in a collection using some
        function ``process_document()``::

            @gen.coroutine
            def process_cursor(cursor):
                while (yield cursor.fetch_next):
                    process_document(document)

            # Get up to 4 cursors.
            cursors = yield collection.parallel_scan(4)
            yield [process_cursor(cursor) for cursor in cursors]

            # All documents have now been processed.

        If ``process_document()`` is a coroutine, do
        ``yield process_document(document)``.

        With :class:`MotorReplicaSetClient`, pass `read_preference` of
        :attr:`~pymongo.read_preference.ReadPreference.SECONDARY_PREFERRED`
        to scan a secondary.

        :Parameters:
          - `num_cursors`: the number of cursors to return

        .. note:: Requires server version **>= 2.5.5**.
        """
        command_cursors = yield self.__parallel_scan(num_cursors, **kwargs)
        motor_command_cursors = [
            MotorCommandCursor(cursor, self)
            for cursor in command_cursors]

        raise gen.Return(motor_command_cursors)

def __init__(self, attr_name, has_write_concern, doc=None):
        """A descriptor that wraps a PyMongo method, such as insert or remove,
        and returns an asynchronous version of the method, which accepts a
        callback or returns a Future.

        :Parameters:
         - `attr_name`: The name of the attribute on the PyMongo class, if
           different from attribute on the Motor class
         - `has_write_concern`: Whether the method accepts getLastError options
        """
        super(Async, self).__init__(doc)
        self.attr_name = attr_name
        self.has_write_concern = has_write_concern

def __init__(self, attr_name, has_write_concern, doc=None):
        """A descriptor that wraps a PyMongo method, such as insert or remove,
        and returns an asynchronous version of the method, which accepts a
        callback or returns a Future.

        :Parameters:
         - `attr_name`: The name of the attribute on the PyMongo class, if
           different from attribute on the Motor class
         - `has_write_concern`: Whether the method accepts getLastError options
        """
        super(Async, self).__init__(doc)
        self.attr_name = attr_name
        self.has_write_concern = has_write_concern

def parallel_scan(self, num_cursors, **kwargs):
        """Scan this entire collection in parallel.

        Returns a list of up to ``num_cursors`` cursors that can be iterated
        concurrently. As long as the collection is not modified during
        scanning, each document appears once in one of the cursors' result
        sets.

        For example, to process each document in a collection using some
        function ``process_document()``::

            @gen.coroutine
            def process_cursor(cursor):
                while (yield cursor.fetch_next):
                    process_document(document)

            # Get up to 4 cursors.
            cursors = yield collection.parallel_scan(4)
            yield [process_cursor(cursor) for cursor in cursors]

            # All documents have now been processed.

        If ``process_document()`` is a coroutine, do
        ``yield process_document(document)``.

        With :class:`MotorReplicaSetClient`, pass `read_preference` of
        :attr:`~pymongo.read_preference.ReadPreference.SECONDARY_PREFERRED`
        to scan a secondary.

        :Parameters:
          - `num_cursors`: the number of cursors to return

        .. note:: Requires server version **>= 2.5.5**.
        """
        command_cursors = yield self.__parallel_scan(num_cursors, **kwargs)
        motor_command_cursors = [
            MotorCommandCursor(cursor, self)
            for cursor in command_cursors]

        raise gen.Return(motor_command_cursors)

def set_read_preference(db):
    if pymongo.version >= "2.1":
        db.read_preference = pymongo.ReadPreference.SECONDARY

def check_compatibility(module, client):
    srv_info = client.server_info()
    if LooseVersion(srv_info['version']) >= LooseVersion('3.0') and LooseVersion(PyMongoVersion) <= LooseVersion('3.0'):
        module.fail_json(msg=' (Note: you must use pymongo 3.0+ with MongoDB >= 3.0)')
    elif LooseVersion(srv_info['version']) >= LooseVersion('2.6') and LooseVersion(PyMongoVersion) <= LooseVersion('2.7'):
        module.fail_json(msg=' (Note: you must use pymongo 2.7.x-2.9.x with MongoDB 2.6)')
    elif LooseVersion(PyMongoVersion) <= LooseVersion('2.5'):
        module.fail_json(msg=' (Note: you must be on mongodb 2.4+ and pymongo 2.5+ to use the roles param)')

def check_dependencies():
    try:
        import nose
        logger.debug("\tNose: %s\n" % str(nose.__version__))
    except ImportError:
        raise ImportError("Nose cannot be imported. Are you sure it's "
                          "installed?")
    try:
        import networkx
        logger.debug("\tnetworkx: %s\n" % str(networkx.__version__))
    except ImportError:
        raise ImportError("Networkx cannot be imported. Are you sure it's "
                          "installed?")
    try:
        import pymongo
        logger.debug("\tpymongo: %s\n" % str(pymongo.version))
        from bson.objectid import ObjectId
    except ImportError:
        raise ImportError("Pymongo cannot be imported. Are you sure it's"
                          " installed?")
    try:
        import numpy
        logger.debug("\tnumpy: %s" % str(numpy.__version__))
    except ImportError:
        raise ImportError("Numpy cannot be imported. Are you sure that it's"
                          " installed?")
    try:
        import scipy
        logger.debug("\tscipy: %s" % str(scipy.__version__))
    except ImportError:
        raise ImportError("Scipy cannot be imported. Are you sure that it's"
                          " installed?")

def check_dependencies():
    try:
        import nose
        logger.debug("\tNose: %s\n" % str(nose.__version__))
    except ImportError:
        raise ImportError("Nose cannot be imported. Are you sure it's "
                          "installed?")
    try:
        import networkx
        logger.debug("\tnetworkx: %s\n" % str(networkx.__version__))
    except ImportError:
        raise ImportError("Networkx cannot be imported. Are you sure it's "
                          "installed?")
    try:
        import pymongo
        logger.debug("\tpymongo: %s\n" % str(pymongo.version))
        from bson.objectid import ObjectId
    except ImportError:
        raise ImportError("Pymongo cannot be imported. Are you sure it's"
                          " installed?")
    try:
        import numpy
        logger.debug("\tnumpy: %s" % str(numpy.__version__))
    except ImportError:
        raise ImportError("Numpy cannot be imported. Are you sure that it's"
                          " installed?")
    try:
        import scipy
        logger.debug("\tscipy: %s" % str(scipy.__version__))
    except ImportError:
        raise ImportError("Scipy cannot be imported. Are you sure that it's"
                          " installed?")

def check_dependencies():
    try:
        import nose
        logger.debug("\tNose: %s\n" % str(nose.__version__))
    except ImportError:
        raise ImportError("Nose cannot be imported. Are you sure it's "
                          "installed?")
    try:
        import networkx
        logger.debug("\tnetworkx: %s\n" % str(networkx.__version__))
    except ImportError:
        raise ImportError("Networkx cannot be imported. Are you sure it's "
                          "installed?")
    try:
        import pymongo
        logger.debug("\tpymongo: %s\n" % str(pymongo.version))
        from bson.objectid import ObjectId
    except ImportError:
        raise ImportError("Pymongo cannot be imported. Are you sure it's"
                          " installed?")
    try:
        import numpy
        logger.debug("\tnumpy: %s" % str(numpy.__version__))
    except ImportError:
        raise ImportError("Numpy cannot be imported. Are you sure that it's"
                          " installed?")
    try:
        import scipy
        logger.debug("\tscipy: %s" % str(scipy.__version__))
    except ImportError:
        raise ImportError("Scipy cannot be imported. Are you sure that it's"
                          " installed?")

def check_compatibility(module, client):
    srv_info = client.server_info()
    if LooseVersion(srv_info['version']) >= LooseVersion('3.2') and LooseVersion(PyMongoVersion) <= LooseVersion('3.2'):
        module.fail_json(msg=' (Note: you must use pymongo 3.2+ with MongoDB >= 3.2)')
    elif LooseVersion(srv_info['version']) >= LooseVersion('3.0') and LooseVersion(PyMongoVersion) <= LooseVersion('2.8'):
        module.fail_json(msg=' (Note: you must use pymongo 2.8+ with MongoDB 3.0)')
    elif LooseVersion(srv_info['version']) >= LooseVersion('2.6') and LooseVersion(PyMongoVersion) <= LooseVersion('2.7'):
        module.fail_json(msg=' (Note: you must use pymongo 2.7+ with MongoDB 2.6)')
    elif LooseVersion(PyMongoVersion) <= LooseVersion('2.5'):
        module.fail_json(msg=' (Note: you must be on mongodb 2.4+ and pymongo 2.5+ to use the roles param)')

def check_compatibility(module, client):
    if LooseVersion(PyMongoVersion) <= LooseVersion('3.0'):
        module.fail_json(msg='Note: you must use pymongo 3.0+')
    srv_info = client.server_info()
    if LooseVersion(srv_info['version']) >= LooseVersion('3.2') and LooseVersion(PyMongoVersion) <= LooseVersion('3.2'):
        module.fail_json(msg=' (Note: you must use pymongo 3.2+ with MongoDB >= 3.2)')

def __init__(self, module):
        self.module = module

        self.login_user     = module.params['login_user']
        self.login_password = module.params['login_password']
        self.login_host     = module.params['login_host']
        self.login_port     = int(module.params['login_port'])
        self.login_database = module.params['login_database']

        self.replica_set = module.params['replica_set']
        self.ssl         = module.params['ssl']

        self.database = module.params['database']

        self.client = self.get_client()

        if self.login_user is None and self.login_password is None:
            if not self.load_mongocnf() and LooseVersion(PyMongoVersion) >= LooseVersion('3.0') and self.database != "admin":
               module.fail_json(msg='The localhost login exception only allows the first admin account to be created')
        elif self.login_password is None or self.login_user is None:
            module.fail_json(msg='when supplying login arguments, both login_user and login_password must be provided')

        if not self.localhost_exception():
            self.client.admin.authenticate(self.login_user, self.login_password, source=self.login_database)

        self.check_compatibility()

def check_compatibility(self):
        srv_info = self.client.server_info()

        if LooseVersion(srv_info['version']) >= LooseVersion('3.2') and LooseVersion(PyMongoVersion) <= LooseVersion('3.2'):
            self.module.fail_json(msg=' (Note: you must use pymongo 3.2+ with MongoDB >= 3.2)')
        elif LooseVersion(srv_info['version']) >= LooseVersion('3.0') and LooseVersion(PyMongoVersion) <= LooseVersion('2.8'):
            self.module.fail_json(msg=' (Note: you must use pymongo 2.8+ with MongoDB 3.0)')
        elif LooseVersion(srv_info['version']) >= LooseVersion('2.6') and LooseVersion(PyMongoVersion) <= LooseVersion('2.7'):
            self.module.fail_json(msg=' (Note: you must use pymongo 2.7+ with MongoDB 2.6)')
        elif LooseVersion(PyMongoVersion) <= LooseVersion('2.5'):
            self.module.fail_json(msg=' (Note: you must be on mongodb 2.4+ and pymongo 2.5+ to use the roles param)')

def localhost_exception(self):
        return self.login_user is None and self.login_password is None \
            and LooseVersion(PyMongoVersion) >= LooseVersion('3.0') and self.database == "admin"