def get_connection_params()

in mysql-connector-python/lib/mysql/connector/django/base.py [0:0]

• 44 lines of code
• 14 McCabe index (conditional complexity)

    def get_connection_params(self) -> Dict[str, Any]:
        kwargs = {
            "charset": "utf8",
            "use_unicode": True,
            "buffered": False,
            "consume_results": True,
        }

        settings_dict = self.settings_dict

        if settings_dict["USER"]:
            kwargs["user"] = settings_dict["USER"]
        if settings_dict["NAME"]:
            kwargs["database"] = settings_dict["NAME"]
        if settings_dict["PASSWORD"]:
            kwargs["passwd"] = settings_dict["PASSWORD"]
        if settings_dict["HOST"].startswith("/"):
            kwargs["unix_socket"] = settings_dict["HOST"]
        elif settings_dict["HOST"]:
            kwargs["host"] = settings_dict["HOST"]
        if settings_dict["PORT"]:
            kwargs["port"] = int(settings_dict["PORT"])
        if settings_dict.get("OPTIONS", {}).get("init_command"):
            kwargs["init_command"] = settings_dict["OPTIONS"]["init_command"]

        # Raise exceptions for database warnings if DEBUG is on
        kwargs["raise_on_warnings"] = settings.DEBUG

        kwargs["client_flags"] = [
            # Need potentially affected rows on UPDATE
            mysql.connector.constants.ClientFlag.FOUND_ROWS,
        ]

        try:
            options = settings_dict["OPTIONS"].copy()
            isolation_level = options.pop("isolation_level", None)
            if isolation_level:
                isolation_level = isolation_level.lower()
                if isolation_level not in self.isolation_levels:
                    valid_levels = ", ".join(
                        f"'{level}'" for level in sorted(self.isolation_levels)
                    )
                    raise ImproperlyConfigured(
                        f"Invalid transaction isolation level '{isolation_level}' "
                        f"specified.\nUse one of {valid_levels}, or None."
                    )
            self.isolation_level = isolation_level
            kwargs.update(options)
        except KeyError:
            # OPTIONS missing is OK
            pass
        return kwargs