Source code for flask_mongoengine.connection

"""Module responsible for connection setup."""
import warnings
from typing import List

import mongoengine

__all__ = (

[docs]def _get_name(setting_name: str) -> str: """ Return known pymongo setting name, or lower case name for unknown. This problem discovered in issue #451. As mentioned there pymongo settings are not case-sensitive, but mongoengine use exact name of some settings for matching, overwriting pymongo behaviour. This function address this issue, and potentially address cases when pymongo will become case-sensitive in some settings by same reasons as mongoengine done. Based on pymongo 4.1.1 settings. """ KNOWN_CAMEL_CASE_SETTINGS = { "directconnection": "directConnection", "maxpoolsize": "maxPoolSize", "minpoolsize": "minPoolSize", "maxidletimems": "maxIdleTimeMS", "maxconnecting": "maxConnecting", "sockettimeoutms": "socketTimeoutMS", "connecttimeoutms": "connectTimeoutMS", "serverselectiontimeoutms": "serverSelectionTimeoutMS", "waitqueuetimeoutms": "waitQueueTimeoutMS", "heartbeatfrequencyms": "heartbeatFrequencyMS", "retrywrites": "retryWrites", "retryreads": "retryReads", "zlibcompressionlevel": "zlibCompressionLevel", "uuidrepresentation": "uuidRepresentation", "srvservicename": "srvServiceName", "wtimeoutms": "wTimeoutMS", "replicaset": "replicaSet", "readpreference": "readPreference", "readpreferencetags": "readPreferenceTags", "maxstalenessseconds": "maxStalenessSeconds", "authsource": "authSource", "authmechanism": "authMechanism", "authmechanismproperties": "authMechanismProperties", "tlsinsecure": "tlsInsecure", "tlsallowinvalidcertificates": "tlsAllowInvalidCertificates", "tlsallowinvalidhostnames": "tlsAllowInvalidHostnames", "tlscafile": "tlsCAFile", "tlscertificatekeyfile": "tlsCertificateKeyFile", "tlscrlfile": "tlsCRLFile", "tlscertificatekeyfilepassword": "tlsCertificateKeyFilePassword", "tlsdisableocspendpointcheck": "tlsDisableOCSPEndpointCheck", "readconcernlevel": "readConcernLevel", } _setting_name = KNOWN_CAMEL_CASE_SETTINGS.get(setting_name.lower()) return setting_name.lower() if _setting_name is None else _setting_name
[docs]def _sanitize_settings(settings: dict) -> dict: """Remove ``MONGODB_`` prefix from dict values, to correct bypass to mongoengine.""" resolved_settings = {} for k, v in settings.items(): # Replace with k.lower().removeprefix("mongodb_") when python 3.8 support ends. key = _get_name(k[8:]) if k.lower().startswith("mongodb_") else _get_name(k) resolved_settings[key] = v return resolved_settings
[docs]def get_connection_settings(config: dict) -> List[dict]: """ Given a config dict, return a sanitized dict of MongoDB connection settings that we can then use to establish connections. For new applications, settings should exist in a ``MONGODB_SETTINGS`` key, but for backward compatibility we also support several config keys prefixed by ``MONGODB_``, e.g. ``MONGODB_HOST``, ``MONGODB_PORT``, etc. """ # If no "MONGODB_SETTINGS", sanitize the "MONGODB_" keys as single connection. if "MONGODB_SETTINGS" not in config: warnings.warn( ( "Passing flat configuration is deprecated. Please check " " " "for more info." ), DeprecationWarning, stacklevel=2, ) config = {k: v for k, v in config.items() if k.lower().startswith("mongodb_")} return [_sanitize_settings(config)] # Sanitize all the settings living under a "MONGODB_SETTINGS" config var settings = config["MONGODB_SETTINGS"] # If MONGODB_SETTINGS is a list of settings dicts, sanitize each dict separately. if isinstance(settings, list): return [_sanitize_settings(settings_dict) for settings_dict in settings] # Otherwise, it should be a single dict describing a single connection. return [_sanitize_settings(settings)]
[docs]def create_connections(config: dict): """ Given Flask application's config dict, extract relevant config vars out of it and establish MongoEngine connection(s) based on them. """ # Validate that the config is a dict and dict is not empty if not config or not isinstance(config, dict): raise TypeError(f"Config dictionary expected, but {type(config)} received.") # Get sanitized connection settings based on the config connection_settings = get_connection_settings(config) connections = {} for connection_setting in connection_settings: alias = connection_setting.setdefault( "alias", mongoengine.DEFAULT_CONNECTION_NAME, ) connection_setting.setdefault("uuidRepresentation", "standard") connections[alias] = mongoengine.connect(**connection_setting) return connections