Function get_connection

mongoengine/connection.py:308–378

Return a connection with a given alias.

(alias=DEFAULT_CONNECTION_NAME, reconnect=False)

def get_connection(alias=DEFAULT_CONNECTION_NAME, reconnect=False):
    """Return a connection with a given alias."""

    # Connect to the database if not already connected
    if reconnect:
        disconnect(alias)

    # If the requested alias already exists in the _connections list, return
    # it immediately.
    if alias in _connections:
        return _connections[alias]

    # Validate that the requested alias exists in the _connection_settings.
    # Raise ConnectionFailure if it doesn't.
    if alias not in _connection_settings:
        if alias == DEFAULT_CONNECTION_NAME:
            msg = "You have not defined a default connection"
        else:
            msg = 'Connection with alias "%s" has not been defined' % alias
        raise ConnectionFailure(msg)

    def _clean_settings(settings_dict):
        if PYMONGO_VERSION < (4,):
            irrelevant_fields_set = {
                "name",
                "username",
                "password",
                "authentication_source",
                "authentication_mechanism",
                "authmechanismproperties",
            }
            rename_fields = {}
        else:
            irrelevant_fields_set = {"name"}
            rename_fields = {
                "authentication_source": "authSource",
                "authentication_mechanism": "authMechanism",
            }
        return {
            rename_fields.get(k, k): v
            for k, v in settings_dict.items()
            if k not in irrelevant_fields_set and v is not None
        }

    raw_conn_settings = _connection_settings[alias].copy()

    # Retrieve a copy of the connection settings associated with the requested
    # alias and remove the database name and authentication info (we don't
    # care about them at this point).
    conn_settings = _clean_settings(raw_conn_settings)
    if DriverInfo is not None:
        conn_settings.setdefault(
            "driver", DriverInfo("MongoEngine", mongoengine.__version__)
        )

    # Determine if we should use PyMongo's or mongomock's MongoClient.
    if "mongo_client_class" in conn_settings:
        mongo_client_class = conn_settings.pop("mongo_client_class")