Implement a single validation function around the service_uri

docs/conf.py

@@ -47,7 +47,7 @@
 autosectionlabel_maxdepth = 2
 
 # AutodocTypehints settings.
-autodoc_typehints = 'description'
+autodoc_typehints = "description"
 always_document_param_types = True
 typehints_defaults = "comma"
 

docs/examples/opentelemetry/main.py

@@ -2,8 +2,8 @@
 
 import time
 
-import valkey
 import uptrace
+import valkey
 from opentelemetry import trace
 from opentelemetry.instrumentation.valkey import ValkeyInstrumentor
 

tests/conftest.py

@@ -9,8 +9,9 @@
 import valkey
 from packaging.version import Version
 from valkey import Sentinel
+from valkey._parsers import parse_url
 from valkey.backoff import NoBackoff
-from valkey.connection import Connection, parse_url
+from valkey.connection import Connection
 from valkey.exceptions import ValkeyClusterException
 from valkey.retry import Retry
 
@@ -275,7 +276,7 @@ def _get_client(
 
     cluster_mode = VALKEY_INFO["cluster_enabled"]
     if not cluster_mode:
-        url_options = parse_url(valkey_url)
+        url_options = parse_url(valkey_url, False)
         url_options.update(kwargs)
         pool = valkey.ConnectionPool(**url_options)
         client = cls(connection_pool=pool)

tests/test_asyncio/conftest.py

@@ -5,9 +5,10 @@
 import pytest_asyncio
 import valkey.asyncio as valkey
 from tests.conftest import VALKEY_INFO
+from valkey._parsers import parse_url
 from valkey.asyncio import Sentinel
 from valkey.asyncio.client import Monitor
-from valkey.asyncio.connection import Connection, parse_url
+from valkey.asyncio.connection import Connection
 from valkey.asyncio.retry import Retry
 from valkey.backoff import NoBackoff
 
@@ -54,7 +55,7 @@ async def client_factory(
         cluster_mode = VALKEY_INFO["cluster_enabled"]
         if not cluster_mode:
             single = kwargs.pop("single_connection_client", False) or single_connection
-            url_options = parse_url(url)
+            url_options = parse_url(url, True)
             url_options.update(kwargs)
             pool = valkey.ConnectionPool(**url_options)
             client = cls(connection_pool=pool)
@@ -269,4 +270,4 @@ def valkey_url(request):
 @pytest.fixture()
 def connect_args(request):
     url = request.config.getoption("--valkey-url")
-    return parse_url(url)
+    return parse_url(url, True)

tests/test_asyncio/test_connection.py

@@ -11,9 +11,10 @@
     _AsyncRESP2Parser,
     _AsyncRESP3Parser,
     _AsyncRESPBase,
+    parse_url,
 )
 from valkey.asyncio import ConnectionPool, Valkey
-from valkey.asyncio.connection import Connection, UnixDomainSocketConnection, parse_url
+from valkey.asyncio.connection import Connection, UnixDomainSocketConnection
 from valkey.asyncio.retry import Retry
 from valkey.backoff import NoBackoff
 from valkey.exceptions import ConnectionError, InvalidResponse, TimeoutError
@@ -300,7 +301,7 @@ async def test_pool_auto_close(request, from_url):
     """Verify that basic Valkey instances have auto_close_connection_pool set to True"""
 
     url: str = request.config.getoption("--valkey-url")
-    url_args = parse_url(url)
+    url_args = parse_url(url, True)
 
     async def get_valkey_connection():
         if from_url:
@@ -342,7 +343,7 @@ async def test_pool_auto_close_disable(request):
     """Verify that auto_close_connection_pool can be disabled (deprecated)"""
 
     url: str = request.config.getoption("--valkey-url")
-    url_args = parse_url(url)
+    url_args = parse_url(url, True)
 
     async def get_valkey_connection():
         url_args["auto_close_connection_pool"] = False
@@ -361,7 +362,7 @@ async def test_valkey_connection_pool(request, from_url):
     have auto_close_connection_pool set to False"""
 
     url: str = request.config.getoption("--valkey-url")
-    url_args = parse_url(url)
+    url_args = parse_url(url, True)
 
     pool = None
 
@@ -393,7 +394,7 @@ async def test_valkey_from_pool(request, from_url):
     have auto_close_connection_pool set to True"""
 
     url: str = request.config.getoption("--valkey-url")
-    url_args = parse_url(url)
+    url_args = parse_url(url, True)
 
     pool = None
 

tests/test_asyncio/test_connection_pool.py

@@ -5,7 +5,8 @@
 import pytest_asyncio
 import valkey.asyncio as valkey
 from tests.conftest import skip_if_server_version_lt
-from valkey.asyncio.connection import Connection, to_bool
+from valkey._parsers.url_parser import to_bool
+from valkey.asyncio.connection import Connection
 
 from .compat import aclosing, mock
 from .conftest import asynccontextmanager
@@ -441,8 +442,8 @@ def test_invalid_scheme_raises_error(self):
         with pytest.raises(ValueError) as cm:
             valkey.ConnectionPool.from_url("localhost")
         assert str(cm.value) == (
-            "Valkey URL must specify one of the following schemes "
-            "(valkey://, valkeys://, unix://)"
+            "Valkey URL must specify one of the following schemes"
+            " ['valkey', 'valkeys', 'redis', 'rediss', 'unix']"
         )
 
 

tests/test_connection.py

@@ -6,14 +6,9 @@
 import pytest
 import valkey
 from valkey import ConnectionPool, Valkey
-from valkey._parsers import _HiredisParser, _RESP2Parser, _RESP3Parser
+from valkey._parsers import _HiredisParser, _RESP2Parser, _RESP3Parser, parse_url
 from valkey.backoff import NoBackoff
-from valkey.connection import (
-    Connection,
-    SSLConnection,
-    UnixDomainSocketConnection,
-    parse_url,
-)
+from valkey.connection import Connection, SSLConnection, UnixDomainSocketConnection
 from valkey.exceptions import ConnectionError, InvalidResponse, TimeoutError
 from valkey.retry import Retry
 from valkey.utils import HIREDIS_AVAILABLE
@@ -222,7 +217,7 @@ def test_pool_auto_close(request, from_url):
     """Verify that basic Valkey instances have auto_close_connection_pool set to True"""
 
     url: str = request.config.getoption("--valkey-url")
-    url_args = parse_url(url)
+    url_args = parse_url(url, False)
 
     def get_valkey_connection():
         if from_url:
@@ -240,7 +235,7 @@ def test_valkey_connection_pool(request, from_url):
     have auto_close_connection_pool set to False"""
 
     url: str = request.config.getoption("--valkey-url")
-    url_args = parse_url(url)
+    url_args = parse_url(url, True)
 
     pool = None
 
@@ -272,7 +267,7 @@ def test_valkey_from_pool(request, from_url):
     have auto_close_connection_pool set to True"""
 
     url: str = request.config.getoption("--valkey-url")
-    url_args = parse_url(url)
+    url_args = parse_url(url, True)
 
     pool = None
 

tests/test_connection_pool.py

@@ -7,7 +7,7 @@
 
 import pytest
 import valkey
-from valkey.connection import to_bool
+from valkey._parsers.url_parser import to_bool
 from valkey.utils import SSL_AVAILABLE
 
 from .conftest import _get_client, skip_if_server_version_lt
@@ -337,15 +337,15 @@ def test_invalid_scheme_raises_error(self):
             valkey.ConnectionPool.from_url("localhost")
         assert str(cm.value) == (
             "Valkey URL must specify one of the following schemes "
-            "(valkey://, valkeys://, unix://)"
+            "['valkey', 'valkeys', 'redis', 'rediss', 'unix']"
         )
 
     def test_invalid_scheme_raises_error_when_double_slash_missing(self):
         with pytest.raises(ValueError) as cm:
             valkey.ConnectionPool.from_url("valkey:foo.bar.com:12345")
         assert str(cm.value) == (
             "Valkey URL must specify one of the following schemes "
-            "(valkey://, valkeys://, unix://)"
+            "['valkey', 'valkeys', 'redis', 'rediss', 'unix']"
         )
 
 

valkey/_parsers/__init__.py

@@ -4,6 +4,7 @@
 from .hiredis import _AsyncHiredisParser, _HiredisParser
 from .resp2 import _AsyncRESP2Parser, _RESP2Parser
 from .resp3 import _AsyncRESP3Parser, _RESP3Parser
+from .url_parser import parse_url
 
 __all__ = [
     "AsyncCommandsParser",
@@ -17,4 +18,5 @@
     "_HiredisParser",
     "_RESP2Parser",
     "_RESP3Parser",
+    "parse_url",
 ]

valkey/_parsers/url_parser.py

@@ -0,0 +1,101 @@
+import re
+from types import MappingProxyType
+from typing import Callable, Mapping, Optional
+from urllib.parse import ParseResult, parse_qs, unquote, urlparse
+
+from valkey.asyncio.connection import ConnectKwargs
+from valkey.asyncio.connection import SSLConnection as SSLConnectionAsync
+from valkey.asyncio.connection import (
+    UnixDomainSocketConnection as UnixDomainSocketConnectionAsync,
+)
+from valkey.connection import SSLConnection, UnixDomainSocketConnection
+
+
+def to_bool(value) -> Optional[bool]:
+    if value is None or value == "":
+        return None
+    if isinstance(value, str) and value.upper() in FALSE_STRINGS:
+        return False
+    return bool(value)
+
+
+FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO")
+
+URL_QUERY_ARGUMENT_PARSERS: Mapping[str, Callable[..., object]] = MappingProxyType(
+    {
+        "db": int,
+        "socket_timeout": float,
+        "socket_connect_timeout": float,
+        "socket_keepalive": to_bool,
+        "retry_on_timeout": to_bool,
+        "max_connections": int,
+        "health_check_interval": int,
+        "ssl_check_hostname": to_bool,
+        "timeout": float,
+    }
+)
+
+
+def parse_url(url: str, async_connection: bool):
+    supported_schemes = ["valkey", "valkeys", "redis", "rediss", "unix"]
+    parsed: ParseResult = urlparse(url)
+    kwargs: ConnectKwargs = {}
+    pattern = re.compile(
+        r"^(?:" + "|".join(map(re.escape, supported_schemes)) + r")://", re.IGNORECASE
+    )
+    if not pattern.match(url):
+        raise ValueError(
+            f"Valkey URL must specify one of the following schemes {supported_schemes}"
+        )
+
+    for name, value_list in parse_qs(parsed.query).items():
+        if value_list and len(value_list) > 0:
+            value = unquote(value_list[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 parsed.username:
+        kwargs["username"] = unquote(parsed.username)
+    if parsed.password:
+        kwargs["password"] = unquote(parsed.password)
+
+    # We only support valkey://, valkeys://, redis://, rediss://, and unix:// schemes.
+    if parsed.scheme == "unix":
+        if parsed.path:
+            kwargs["path"] = unquote(parsed.path)
+        kwargs["connection_class"] = (
+            UnixDomainSocketConnectionAsync
+            if async_connection
+            else UnixDomainSocketConnection
+        )
+
+    elif parsed.scheme in supported_schemes:
+        if parsed.hostname:
+            kwargs["host"] = unquote(parsed.hostname)
+        if parsed.port:
+            kwargs["port"] = int(parsed.port)
+
+        # If there's a path argument, use it as the db argument if a
+        # querystring value wasn't specified
+        if parsed.path and "db" not in kwargs:
+            try:
+                kwargs["db"] = int(unquote(parsed.path).replace("/", ""))
+            except (AttributeError, ValueError):
+                pass
+
+        if parsed.scheme in ("valkeys", "rediss"):
+            kwargs["connection_class"] = (
+                SSLConnectionAsync if async_connection else SSLConnection
+            )
+    else:
+        raise ValueError(
+            f"Valkey URL must specify one of the following schemes {supported_schemes}"
+        )
+
+    return kwargs

valkey/asyncio/cluster.py

@@ -25,19 +25,14 @@
     DEFAULT_EVICTION_POLICY,
     AbstractCache,
 )
-from valkey._parsers import AsyncCommandsParser, Encoder
+from valkey._parsers import AsyncCommandsParser, Encoder, parse_url
 from valkey._parsers.helpers import (
     _ValkeyCallbacks,
     _ValkeyCallbacksRESP2,
     _ValkeyCallbacksRESP3,
 )
 from valkey.asyncio.client import ResponseCallbackT
-from valkey.asyncio.connection import (
-    Connection,
-    DefaultParser,
-    SSLConnection,
-    parse_url,
-)
+from valkey.asyncio.connection import Connection, DefaultParser, SSLConnection
 from valkey.asyncio.lock import Lock
 from valkey.asyncio.retry import Retry
 from valkey.backoff import default_backoff
@@ -211,7 +206,7 @@ def from_url(cls, url: str, **kwargs: Any) -> "ValkeyCluster":
         :class:`~valkey.asyncio.connection.Connection` when created.
         In the case of conflicting arguments, querystring arguments are used.
         """
-        kwargs.update(parse_url(url))
+        kwargs.update(parse_url(url, True))
         if kwargs.pop("connection_class", None) is SSLConnection:
             kwargs["ssl"] = True
         return cls(**kwargs)