def _validate_connection_attributes()

in mysqlx-connector-python/lib/mysqlx/connection.py [0:0]

• 92 lines of code
• 34 McCabe index (conditional complexity)

def _validate_connection_attributes(settings: Dict[str, Any]) -> None:
    """Validate connection-attributes.

    Args:
        settings (dict): Settings dictionary.

    Raises:
        :class:`mysqlx.InterfaceError`: If attribute name or value exceeds size.
    """
    attributes = {}
    if "connection-attributes" not in settings:
        return

    conn_attrs = settings["connection-attributes"]

    if isinstance(conn_attrs, str):
        if conn_attrs == "":
            settings["connection-attributes"] = {}
            return
        if not (
            conn_attrs.startswith("[") and conn_attrs.endswith("]")
        ) and conn_attrs not in ["False", "false", "True", "true"]:
            raise InterfaceError(
                "The value of 'connection-attributes' must be a boolean or a "
                f"list of key-value pairs, found: '{conn_attrs}'"
            )
        if conn_attrs in ["False", "false", "True", "true"]:
            if conn_attrs in ["False", "false"]:
                settings["connection-attributes"] = False
            else:
                settings["connection-attributes"] = {}
            return
        conn_attributes = conn_attrs[1:-1].split(",")
        for attr in conn_attributes:
            if attr == "":
                continue
            attr_name_val = attr.split("=")
            attr_name = attr_name_val[0]
            attr_val = attr_name_val[1] if len(attr_name_val) > 1 else ""
            if attr_name in attributes:
                raise InterfaceError(
                    f"Duplicate key '{attr_name}' used in connection-attributes"
                )
            attributes[attr_name] = attr_val
    elif isinstance(conn_attrs, dict):
        for attr_name in conn_attrs:
            attr_value = conn_attrs[attr_name]
            if not isinstance(attr_value, str):
                attr_value = repr(attr_value)
            attributes[attr_name] = attr_value
    elif isinstance(conn_attrs, bool) or conn_attrs in [0, 1]:
        if conn_attrs:
            settings["connection-attributes"] = {}
        else:
            settings["connection-attributes"] = False
        return
    elif isinstance(conn_attrs, set):
        for attr_name in conn_attrs:
            attributes[attr_name] = ""
    elif isinstance(conn_attrs, list):
        for attr in conn_attrs:
            if attr == "":
                continue
            attr_name_val = attr.split("=")
            attr_name = attr_name_val[0]
            attr_val = attr_name_val[1] if len(attr_name_val) > 1 else ""
            if attr_name in attributes:
                raise InterfaceError(
                    f"Duplicate key '{attr_name}' used in connection-attributes"
                )
            attributes[attr_name] = attr_val
    elif not isinstance(conn_attrs, bool):
        raise InterfaceError(
            "connection-attributes must be Boolean or a list of key-value "
            f"pairs, found: '{conn_attrs}'"
        )

    if attributes:
        for attr_name, attr_value in attributes.items():
            # Validate name type
            if not isinstance(attr_name, str):
                raise InterfaceError(
                    f"Attribute name '{attr_name}' must be a string type"
                )
            # Validate attribute name limit 32 characters
            if len(attr_name) > 32:
                raise InterfaceError(
                    f"Attribute name '{attr_name}' exceeds 32 characters limit size"
                )
            # Validate names in connection-attributes cannot start with "_"
            if attr_name.startswith("_"):
                raise InterfaceError(
                    "Key names in connection-attributes cannot start with "
                    f"'_', found: '{attr_name}'"
                )

            # Validate value type
            if not isinstance(attr_value, str):
                raise InterfaceError(
                    f"Attribute '{attr_name}' value: '{attr_value}' must be "
                    "a string type"
                )
            # Validate attribute value limit 1024 characters
            if len(attr_value) > 1024:
                raise InterfaceError(
                    f"Attribute '{attr_name}' value: '{attr_value}' exceeds "
                    "1024 characters limit size"
                )

    settings["connection-attributes"] = attributes