def _parse_address_list()

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

• 37 lines of code
• 13 McCabe index (conditional complexity)

def _parse_address_list(path: str) -> Union[Dict[str, List[Dict]], Dict]:
    """Parses a list of host, port pairs.

    Args:
        path: String containing a list of routers or just router

    Returns:
        Returns a dict with parsed values of host, port and priority if
        specified.
    """
    path = path.replace(" ", "")
    array = (
        not ("," not in path and path.count(":") > 1 and path.count("[") == 1)
        and path.startswith("[")
        and path.endswith("]")
    )

    routers = []
    address_list = _SPLIT_RE.split(path[1:-1] if array else path)
    priority_count = 0
    for address in address_list:
        router: Dict = {}

        match = _PRIORITY_RE.match(address)
        if match:
            address = match.group(1)
            router["priority"] = int(match.group(2))
            priority_count += 1
        else:
            match = _ROUTER_RE.match(address)
            if match:
                address = match.group(1)
                router["priority"] = 100

        match = urlparse(f"//{address}")
        if not match.hostname:
            raise InterfaceError(f"Invalid address: {address}")

        try:
            router.update(host=match.hostname, port=match.port)
        except ValueError as err:
            raise ProgrammingError(f"Invalid URI: {err}", 4002) from err

        routers.append(router)

    if 0 < priority_count < len(address_list):
        raise ProgrammingError(
            "You must either assign no priority to any of the routers or give "
            "a priority for every router",
            4000,
        )

    return {"routers": routers} if array else routers[0]