def parse_url()

in redis/connection.py [0:0]

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

def parse_url(url):
    if not (
        url.startswith("redis://")
        or url.startswith("rediss://")
        or url.startswith("unix://")
    ):
        raise ValueError(
            "Redis URL must specify one of the following "
            "schemes (redis://, rediss://, unix://)"
        )

    url = urlparse(url)
    kwargs = {}

    for name, value in parse_qs(url.query).items():
        if value and len(value) > 0:
            value = unquote(value[0])
            parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
            if parser:
                try:
                    kwargs[name] = parser(value)
                except (TypeError, ValueError):
                    raise ValueError(f"Invalid value for '{name}' in connection URL.")
            else:
                kwargs[name] = value

    if url.username:
        kwargs["username"] = unquote(url.username)
    if url.password:
        kwargs["password"] = unquote(url.password)

    # We only support redis://, rediss:// and unix:// schemes.
    if url.scheme == "unix":
        if url.path:
            kwargs["path"] = unquote(url.path)
        kwargs["connection_class"] = UnixDomainSocketConnection

    else:  # implied:  url.scheme in ("redis", "rediss"):
        if url.hostname:
            kwargs["host"] = unquote(url.hostname)
        if url.port:
            kwargs["port"] = int(url.port)

        # If there's a path argument, use it as the db argument if a
        # querystring value wasn't specified
        if url.path and "db" not in kwargs:
            try:
                kwargs["db"] = int(unquote(url.path).replace("/", ""))
            except (AttributeError, ValueError):
                pass

        if url.scheme == "rediss":
            kwargs["connection_class"] = SSLConnection

    return kwargs