Avoid unclean interpreter shutdown

src/neo4j/__init__.py

@@ -176,8 +176,7 @@ def __getattr__(name):
     if name in (
         "log", "Config", "PoolConfig", "SessionConfig", "WorkspaceConfig"
     ):
-        from ._meta import deprecation_warn
-        deprecation_warn(
+        _deprecation_warn(
             "Importing {} from neo4j is deprecated without replacement. It's "
             "internal and will be removed in a future version."
             .format(name),

src/neo4j/_async/io/_bolt.py

@@ -37,6 +37,7 @@
 from ..._meta import USER_AGENT
 from ...addressing import ResolvedAddress
 from ...api import (
+    Auth,
     ServerInfo,
     Version,
 )
@@ -178,7 +179,6 @@ def _to_auth_dict(cls, auth):
         if not auth:
             return {}
         elif isinstance(auth, tuple) and 2 <= len(auth) <= 3:
-            from ...api import Auth
             return vars(Auth("basic", *auth))
         else:
             try:
@@ -837,8 +837,7 @@ async def _set_defunct_write(self, error=None, silent=False):
         await self._set_defunct(message, error=error, silent=silent)
 
     async def _set_defunct(self, message, error=None, silent=False):
-        from ._pool import AsyncBoltPool
-        direct_driver = isinstance(self.pool, AsyncBoltPool)
+        direct_driver = getattr(self.pool, "is_direct_pool", False)
         user_cancelled = isinstance(error, asyncio.CancelledError)
 
         if error:

src/neo4j/_async/io/_pool.py

@@ -100,6 +100,11 @@ def __init__(self, opener, pool_config, workspace_config):
         self.lock = AsyncCooperativeRLock()
         self.cond = AsyncCondition(self.lock)
 
+    @property
+    @abc.abstractmethod
+    def is_direct_pool(self) -> bool:
+        ...
+
     async def __aenter__(self):
         return self
 
@@ -490,6 +495,8 @@ async def close(self):
 
 class AsyncBoltPool(AsyncIOPool):
 
+    is_direct_pool = True
+
     @classmethod
     def open(cls, address, *, pool_config, workspace_config):
         """Create a new BoltPool
@@ -536,6 +543,8 @@ class AsyncNeo4jPool(AsyncIOPool):
     """ Connection pool with routing table.
     """
 
+    is_direct_pool = False
+
     @classmethod
     def open(cls, *addresses, pool_config, workspace_config,
              routing_context=None):
@@ -578,6 +587,7 @@ def __init__(self, opener, pool_config, workspace_config, address):
         self.address = address
         self.routing_tables = {}
         self.refresh_lock = AsyncRLock()
+        self.is_direct_pool = False
 
     def __repr__(self):
         """ The representation shows the initial routing addresses.

src/neo4j/_sync/io/_bolt.py

@@ -37,6 +37,7 @@
 from ..._meta import USER_AGENT
 from ...addressing import ResolvedAddress
 from ...api import (
+    Auth,
     ServerInfo,
     Version,
 )
@@ -178,7 +179,6 @@ def _to_auth_dict(cls, auth):
         if not auth:
             return {}
         elif isinstance(auth, tuple) and 2 <= len(auth) <= 3:
-            from ...api import Auth
             return vars(Auth("basic", *auth))
         else:
             try:
@@ -837,8 +837,7 @@ def _set_defunct_write(self, error=None, silent=False):
         self._set_defunct(message, error=error, silent=silent)
 
     def _set_defunct(self, message, error=None, silent=False):
-        from ._pool import BoltPool
-        direct_driver = isinstance(self.pool, BoltPool)
+        direct_driver = getattr(self.pool, "is_direct_pool", False)
         user_cancelled = isinstance(error, asyncio.CancelledError)
 
         if error:

src/neo4j/_sync/io/_pool.py

@@ -97,6 +97,11 @@ def __init__(self, opener, pool_config, workspace_config):
         self.lock = CooperativeRLock()
         self.cond = Condition(self.lock)
 
+    @property
+    @abc.abstractmethod
+    def is_direct_pool(self) -> bool:
+        ...
+
     def __enter__(self):
         return self
 
@@ -487,6 +492,8 @@ def close(self):
 
 class BoltPool(IOPool):
 
+    is_direct_pool = True
+
     @classmethod
     def open(cls, address, *, pool_config, workspace_config):
         """Create a new BoltPool
@@ -533,6 +540,8 @@ class Neo4jPool(IOPool):
     """ Connection pool with routing table.
     """
 
+    is_direct_pool = False
+
     @classmethod
     def open(cls, *addresses, pool_config, workspace_config,
              routing_context=None):
@@ -575,6 +584,7 @@ def __init__(self, opener, pool_config, workspace_config, address):
         self.address = address
         self.routing_tables = {}
         self.refresh_lock = RLock()
+        self.is_direct_pool = False
 
     def __repr__(self):
         """ The representation shows the initial routing addresses.

src/neo4j/time/_clock_implementations.py

@@ -24,6 +24,7 @@
     Structure,
 )
 from platform import uname
+from time import time
 
 from . import (
     Clock,
@@ -53,7 +54,6 @@ def available(cls):
         return True
 
     def utc_time(self):
-        from time import time
         seconds, nanoseconds = nano_divmod(int(time() * 1000000), 1000000)
         return ClockTime(seconds, nanoseconds * 1000)
 

testkitbackend/_async/requests.py

@@ -19,6 +19,7 @@
 import datetime
 import json
 import re
+import ssl
 import warnings
 from os import path
 
@@ -56,7 +57,6 @@ def load_config():
         config = json.load(fd)
     skips = config["skips"]
     features = [k for k, v in config["features"].items() if v is True]
-    import ssl
     if ssl.HAS_TLSv1_3:
         features += ["Feature:TLS:1.3"]
     return skips, features

testkitbackend/_sync/requests.py

@@ -19,6 +19,7 @@
 import datetime
 import json
 import re
+import ssl
 import warnings
 from os import path
 
@@ -56,7 +57,6 @@ def load_config():
         config = json.load(fd)
     skips = config["skips"]
     features = [k for k, v in config["features"].items() if v is True]
-    import ssl
     if ssl.HAS_TLSv1_3:
         features += ["Feature:TLS:1.3"]
     return skips, features

tests/conftest.py

@@ -17,21 +17,17 @@
 
 
 import asyncio
-import warnings
+import sys
 from functools import wraps
-from os import environ
 
 import pytest
 import pytest_asyncio
 
 from neo4j import (
     AsyncGraphDatabase,
-    ExperimentalWarning,
     GraphDatabase,
 )
-from neo4j._exceptions import BoltHandshakeError
-from neo4j._sync.io import Bolt
-from neo4j.exceptions import ServiceUnavailable
+from neo4j.debug import watch
 
 from . import env
 
@@ -189,8 +185,5 @@ def _():
 
 @pytest.fixture
 def watcher():
-    import sys
-
-    from neo4j.debug import watch
     with watch("neo4j", out=sys.stdout, colour=True):
         yield

tests/unit/async_/io/test_direct.py

@@ -85,6 +85,8 @@ def timedout(self):
 
 
 class AsyncFakeBoltPool(AsyncIOPool):
+    is_direct_pool = False
+
     def __init__(self, address, *, auth=None, **config):
         config["auth"] = static_auth(None)
         self.pool_config, self.workspace_config = Config.consume_chain(config, PoolConfig, WorkspaceConfig)

tests/unit/common/test_api.py

@@ -24,6 +24,7 @@
 import pytest
 
 import neo4j.api
+from neo4j.addressing import Address
 from neo4j.exceptions import ConfigurationError
 
 
@@ -276,9 +277,6 @@ def test_version_to_bytes_with_valid_bolt_version(
 
 
 def test_serverinfo_initialization() -> None:
-
-    from neo4j.addressing import Address
-
     address = Address(("bolt://localhost", 7687))
     version = neo4j.Version(3, 0)
 
@@ -301,8 +299,6 @@ def test_serverinfo_initialization() -> None:
 def test_serverinfo_with_metadata(
     test_input, expected_agent, protocol_version
 ) -> None:
-    from neo4j.addressing import Address
-
     address = Address(("bolt://localhost", 7687))
     version = neo4j.Version(*protocol_version)
 

tests/unit/sync/io/test_direct.py

@@ -85,6 +85,8 @@ def timedout(self):
 
 
 class FakeBoltPool(IOPool):
+    is_direct_pool = False
+
     def __init__(self, address, *, auth=None, **config):
         config["auth"] = static_auth(None)
         self.pool_config, self.workspace_config = Config.consume_chain(config, PoolConfig, WorkspaceConfig)