Make internal config parsing helper functions private

CHANGELOG.md

@@ -61,13 +61,16 @@ See also https://github.com/neo4j/neo4j-python-driver/wiki for a full changelog.
   - Remove `ExperimentalWarning` and turn the few left instances of it into `PreviewWarning`.
   - Deprecate importing `PreviewWarning` from `neo4j`.  
     Import it from `neo4j.warnings` instead.
-- Make undocumented internal constants private:
+- Make undocumented internal constants and helper functions private:
   - `neo4j.api`
     - `DRIVER_BOLT`
     - `DRIVER_NEO4J`
     - `SECURITY_TYPE_NOT_SECURE`
     - `SECURITY_TYPE_SECURE`
     - `SECURITY_TYPE_SELF_SIGNED_CERTIFICATE`
+    - `check_access_mode`
+    - `parse_neo4j_uri`
+    - `parse_routing_context`
   - `neo4j.exceptions`
     - `CLASSIFICATION_CLIENT`
     - `CLASSIFICATION_DATABASE`

src/neo4j/_api.py

@@ -18,6 +18,13 @@
 
 import typing as t
 from enum import Enum
+from urllib.parse import (
+    parse_qs,
+    urlparse,
+)
+
+from . import api
+from .exceptions import ConfigurationError
 
 
 if t.TYPE_CHECKING:
@@ -38,6 +45,9 @@
     "NotificationSeverity",
     "RoutingControl",
     "TelemetryAPI",
+    "check_access_mode",
+    "parse_neo4j_uri",
+    "parse_routing_context",
 ]
 
 
@@ -51,6 +61,96 @@
 SECURITY_TYPE_SECURE: te.Final[str] = "SECURITY_TYPE_SECURE"
 
 
+def parse_neo4j_uri(uri):
+    parsed = urlparse(uri)
+
+    if parsed.username:
+        raise ConfigurationError("Username is not supported in the URI")
+
+    if parsed.password:
+        raise ConfigurationError("Password is not supported in the URI")
+
+    if parsed.scheme == api.URI_SCHEME_BOLT_ROUTING:
+        raise ConfigurationError(
+            f"Uri scheme {parsed.scheme!r} has been renamed. "
+            f"Use {api.URI_SCHEME_NEO4J!r}"
+        )
+    elif parsed.scheme == api.URI_SCHEME_BOLT:
+        driver_type = DRIVER_BOLT
+        security_type = SECURITY_TYPE_NOT_SECURE
+    elif parsed.scheme == api.URI_SCHEME_BOLT_SELF_SIGNED_CERTIFICATE:
+        driver_type = DRIVER_BOLT
+        security_type = SECURITY_TYPE_SELF_SIGNED_CERTIFICATE
+    elif parsed.scheme == api.URI_SCHEME_BOLT_SECURE:
+        driver_type = DRIVER_BOLT
+        security_type = SECURITY_TYPE_SECURE
+    elif parsed.scheme == api.URI_SCHEME_NEO4J:
+        driver_type = DRIVER_NEO4J
+        security_type = SECURITY_TYPE_NOT_SECURE
+    elif parsed.scheme == api.URI_SCHEME_NEO4J_SELF_SIGNED_CERTIFICATE:
+        driver_type = DRIVER_NEO4J
+        security_type = SECURITY_TYPE_SELF_SIGNED_CERTIFICATE
+    elif parsed.scheme == api.URI_SCHEME_NEO4J_SECURE:
+        driver_type = DRIVER_NEO4J
+        security_type = SECURITY_TYPE_SECURE
+    else:
+        supported_schemes = [
+            api.URI_SCHEME_BOLT,
+            api.URI_SCHEME_BOLT_SELF_SIGNED_CERTIFICATE,
+            api.URI_SCHEME_BOLT_SECURE,
+            api.URI_SCHEME_NEO4J,
+            api.URI_SCHEME_NEO4J_SELF_SIGNED_CERTIFICATE,
+            api.URI_SCHEME_NEO4J_SECURE,
+        ]
+        raise ConfigurationError(
+            f"URI scheme {parsed.scheme!r} is not supported. "
+            f"Supported URI schemes are {supported_schemes}. "
+            "Examples: bolt://host[:port] or "
+            "neo4j://host[:port][?routing_context]"
+        )
+
+    return driver_type, security_type, parsed
+
+
+def check_access_mode(access_mode):
+    if access_mode not in {api.READ_ACCESS, api.WRITE_ACCESS}:
+        raise ValueError(
+            f"Unsupported access mode {access_mode}, must be one of "
+            f"'{api.READ_ACCESS}' or '{api.WRITE_ACCESS}'."
+        )
+
+    return access_mode
+
+
+def parse_routing_context(query):
+    """
+    Parse the query portion of a URI.
+
+    Generates a routing context dictionary.
+    """
+    if not query:
+        return {}
+
+    context = {}
+    parameters = parse_qs(query, True)
+    for key in parameters:
+        value_list = parameters[key]
+        if len(value_list) != 1:
+            raise ConfigurationError(
+                f"Duplicated query parameters with key '{key}', value "
+                f"'{value_list}' found in query string '{query}'"
+            )
+        value = value_list[0]
+        if not value:
+            raise ConfigurationError(
+                f"Invalid parameters:'{key}={value}' in query string "
+                f"'{query}'."
+            )
+        context[key] = value
+
+    return context
+
+
 class NotificationMinimumSeverity(str, Enum):
     """
     Filter notifications returned by the server by minimum severity.

src/neo4j/_async/driver.py

@@ -34,6 +34,8 @@
     DRIVER_BOLT,
     DRIVER_NEO4J,
     NotificationMinimumSeverity,
+    parse_neo4j_uri,
+    parse_routing_context,
     RoutingControl,
     SECURITY_TYPE_SECURE,
     SECURITY_TYPE_SELF_SIGNED_CERTIFICATE,
@@ -64,8 +66,6 @@
     Auth,
     BookmarkManager,
     Bookmarks,
-    parse_neo4j_uri,
-    parse_routing_context,
     READ_ACCESS,
     ServerInfo,
     TRUST_ALL_CERTIFICATES,

src/neo4j/_async/io/_pool.py

@@ -31,6 +31,7 @@
 from logging import getLogger
 from random import choice
 
+from ..._api import check_access_mode
 from ..._async_compat.concurrency import (
     AsyncCondition,
     AsyncCooperativeRLock,
@@ -45,10 +46,7 @@
 )
 from ..._exceptions import BoltError
 from ..._routing import RoutingTable
-from ...api import (
-    check_access_mode,
-    READ_ACCESS,
-)
+from ...api import READ_ACCESS
 from ...exceptions import (
     ConfigurationError,
     ConnectionAcquisitionTimeoutError,

src/neo4j/api.py

@@ -20,13 +20,6 @@
 
 import abc
 import typing as t
-from urllib.parse import (
-    parse_qs,
-    urlparse,
-)
-
-from . import _api
-from .exceptions import ConfigurationError
 
 
 if t.TYPE_CHECKING:
@@ -65,11 +58,8 @@
     "ServerInfo",
     "basic_auth",
     "bearer_auth",
-    "check_access_mode",
     "custom_auth",
     "kerberos_auth",
-    "parse_neo4j_uri",
-    "parse_routing_context",
 ]
 
 
@@ -443,96 +433,3 @@ async def update_bookmarks(
     async def get_bookmarks(self) -> t.Collection[str]: ...
 
     get_bookmarks.__doc__ = BookmarkManager.get_bookmarks.__doc__
-
-
-# TODO: 6.0 - make this function private
-def parse_neo4j_uri(uri):
-    parsed = urlparse(uri)
-
-    if parsed.username:
-        raise ConfigurationError("Username is not supported in the URI")
-
-    if parsed.password:
-        raise ConfigurationError("Password is not supported in the URI")
-
-    if parsed.scheme == URI_SCHEME_BOLT_ROUTING:
-        raise ConfigurationError(
-            f"Uri scheme {parsed.scheme!r} has been renamed. "
-            f"Use {URI_SCHEME_NEO4J!r}"
-        )
-    elif parsed.scheme == URI_SCHEME_BOLT:
-        driver_type = _api.DRIVER_BOLT
-        security_type = _api.SECURITY_TYPE_NOT_SECURE
-    elif parsed.scheme == URI_SCHEME_BOLT_SELF_SIGNED_CERTIFICATE:
-        driver_type = _api.DRIVER_BOLT
-        security_type = _api.SECURITY_TYPE_SELF_SIGNED_CERTIFICATE
-    elif parsed.scheme == URI_SCHEME_BOLT_SECURE:
-        driver_type = _api.DRIVER_BOLT
-        security_type = _api.SECURITY_TYPE_SECURE
-    elif parsed.scheme == URI_SCHEME_NEO4J:
-        driver_type = _api.DRIVER_NEO4J
-        security_type = _api.SECURITY_TYPE_NOT_SECURE
-    elif parsed.scheme == URI_SCHEME_NEO4J_SELF_SIGNED_CERTIFICATE:
-        driver_type = _api.DRIVER_NEO4J
-        security_type = _api.SECURITY_TYPE_SELF_SIGNED_CERTIFICATE
-    elif parsed.scheme == URI_SCHEME_NEO4J_SECURE:
-        driver_type = _api.DRIVER_NEO4J
-        security_type = _api.SECURITY_TYPE_SECURE
-    else:
-        supported_schemes = [
-            URI_SCHEME_BOLT,
-            URI_SCHEME_BOLT_SELF_SIGNED_CERTIFICATE,
-            URI_SCHEME_BOLT_SECURE,
-            URI_SCHEME_NEO4J,
-            URI_SCHEME_NEO4J_SELF_SIGNED_CERTIFICATE,
-            URI_SCHEME_NEO4J_SECURE,
-        ]
-        raise ConfigurationError(
-            f"URI scheme {parsed.scheme!r} is not supported. "
-            f"Supported URI schemes are {supported_schemes}. "
-            "Examples: bolt://host[:port] or "
-            "neo4j://host[:port][?routing_context]"
-        )
-
-    return driver_type, security_type, parsed
-
-
-# TODO: 6.0 - make this function private
-def check_access_mode(access_mode):
-    if access_mode not in {READ_ACCESS, WRITE_ACCESS}:
-        raise ValueError(
-            f"Unsupported access mode {access_mode}, must be one of "
-            f"'{READ_ACCESS}' or '{WRITE_ACCESS}'."
-        )
-
-    return access_mode
-
-
-# TODO: 6.0 - make this function private
-def parse_routing_context(query):
-    """
-    Parse the query portion of a URI.
-
-    Generates a routing context dictionary.
-    """
-    if not query:
-        return {}
-
-    context = {}
-    parameters = parse_qs(query, True)
-    for key in parameters:
-        value_list = parameters[key]
-        if len(value_list) != 1:
-            raise ConfigurationError(
-                f"Duplicated query parameters with key '{key}', value "
-                f"'{value_list}' found in query string '{query}'"
-            )
-        value = value_list[0]
-        if not value:
-            raise ConfigurationError(
-                f"Invalid parameters:'{key}={value}' in query string "
-                f"'{query}'."
-            )
-        context[key] = value
-
-    return context

tests/unit/common/test_api.py

@@ -230,25 +230,25 @@ def test_uri_scheme(
 ) -> None:
     if expected_error:
         with pytest.raises(expected_error):
-            neo4j.api.parse_neo4j_uri(test_input)
+            neo4j._api.parse_neo4j_uri(test_input)
     else:
-        driver_type, security_type, _parsed = neo4j.api.parse_neo4j_uri(
+        driver_type, security_type, _parsed = neo4j._api.parse_neo4j_uri(
             test_input
         )
         assert driver_type == expected_driver_type
         assert security_type == expected_security_type
 
 
 def test_parse_routing_context() -> None:
-    context = neo4j.api.parse_routing_context(query="name=molly&color=white")
+    context = neo4j._api.parse_routing_context(query="name=molly&color=white")
     assert context == {"name": "molly", "color": "white"}
 
 
 def test_parse_routing_context_should_error_when_value_missing() -> None:
     with pytest.raises(ConfigurationError):
-        neo4j.api.parse_routing_context("name=&color=white")
+        neo4j._api.parse_routing_context("name=&color=white")
 
 
 def test_parse_routing_context_should_error_when_key_duplicate() -> None:
     with pytest.raises(ConfigurationError):
-        neo4j.api.parse_routing_context("name=molly&name=white")
+        neo4j._api.parse_routing_context("name=molly&name=white")