Mitigate errors during resource warning on interpreter shutdown

src/neo4j/_async/driver.py

@@ -503,20 +503,27 @@ async def __aenter__(self) -> AsyncDriver:
     async def __aexit__(self, exc_type, exc_value, traceback):
         await self.close()
 
-    def __del__(self):
+    # Copy globals as function locals to make sure that they are available
+    # during Python shutdown when the Pool is destroyed.
+    def __del__(
+        self, _unclosed_resource_warn=unclosed_resource_warn,
+        _is_async_code=AsyncUtil.is_async_code,
+        _deprecation_warn=deprecation_warn,
+    ):
         if not self._closed:
-            unclosed_resource_warn(self)
+            _unclosed_resource_warn(self)
         # TODO: 6.0 - remove this
+        if _is_async_code:
+            return
         if not self._closed:
-            if not AsyncUtil.is_async_code:
-                deprecation_warn(
-                    "Relying on AsyncDriver's destructor to close the session "
-                    "is deprecated. Please make sure to close the session. "
-                    "Use it as a context (`with` statement) or make sure to "
-                    "call `.close()` explicitly. Future versions of the "
-                    "driver will not close drivers automatically."
-                )
-                self.close()
+            _deprecation_warn(
+                "Relying on AsyncDriver's destructor to close the session "
+                "is deprecated. Please make sure to close the session. "
+                "Use it as a context (`with` statement) or make sure to "
+                "call `.close()` explicitly. Future versions of the "
+                "driver will not close drivers automatically."
+            )
+            self.close()
 
     def _check_state(self):
         if self._closed:

src/neo4j/_async/work/workspace.py

@@ -59,15 +59,19 @@ def __init__(self, pool, config):
         self._closed = False
         super().__init__()
 
-    def __del__(self):
+    def __del__(
+        self, _unclosed_resource_warn=unclosed_resource_warn,
+        _is_async_code=AsyncUtil.is_async_code,
+        _deprecation_warn=deprecation_warn,
+    ):
         if self._closed:
             return
-        unclosed_resource_warn(self)
+        _unclosed_resource_warn(self)
         # TODO: 6.0 - remove this
-        if asyncio.iscoroutinefunction(self.close):
+        if _is_async_code:
             return
         try:
-            deprecation_warn(
+            _deprecation_warn(
                 "Relying on AsyncSession's destructor to close the session "
                 "is deprecated. Please make sure to close the session. Use it "
                 "as a context (`with` statement) or make sure to call "

src/neo4j/_meta.py

@@ -19,7 +19,6 @@
 import asyncio
 import platform
 import sys
-import tracemalloc
 import typing as t
 from functools import wraps
 from inspect import isclass
@@ -91,8 +90,11 @@ def copy_signature(_: _FuncT) -> t.Callable[[t.Callable], _FuncT]:
     return _id
 
 
-def deprecation_warn(message, stack_level=1):
-    warn(message, category=DeprecationWarning, stacklevel=stack_level + 1)
+
+# Copy globals as function locals to make sure that they are available
+# during Python shutdown when the Pool is destroyed.
+def deprecation_warn(message, stack_level=1, _warn=warn):
+    _warn(message, category=DeprecationWarning, stacklevel=stack_level + 1)
 
 
 def deprecated(message: str) -> t.Callable[[_FuncT], _FuncT]:
@@ -224,12 +226,9 @@ def inner(*args, **kwargs):
     return decorator
 
 
-def unclosed_resource_warn(obj):
-    msg = f"Unclosed {obj!r}."
-    trace = tracemalloc.get_object_traceback(obj)
-    if trace:
-        msg += "\nObject allocated at (most recent call last):\n"
-        msg += "\n".join(trace.format())
-    else:
-        msg += "\nEnable tracemalloc to get the object allocation traceback."
-    warn(msg, ResourceWarning, stacklevel=2, source=obj)
+# Copy globals as function locals to make sure that they are available
+# during Python shutdown when the Pool is destroyed.
+def unclosed_resource_warn(obj, _warn=warn):
+    cls_name = obj.__class__.__name__
+    msg = f"unclosed  {cls_name}: {obj!r}."
+    _warn(msg, ResourceWarning, stacklevel=2, source=obj)

src/neo4j/_sync/driver.py

@@ -502,20 +502,27 @@ def __enter__(self) -> Driver:
     def __exit__(self, exc_type, exc_value, traceback):
         self.close()
 
-    def __del__(self):
+    # Copy globals as function locals to make sure that they are available
+    # during Python shutdown when the Pool is destroyed.
+    def __del__(
+        self, _unclosed_resource_warn=unclosed_resource_warn,
+        _is_async_code=Util.is_async_code,
+        _deprecation_warn=deprecation_warn,
+    ):
         if not self._closed:
-            unclosed_resource_warn(self)
+            _unclosed_resource_warn(self)
         # TODO: 6.0 - remove this
+        if _is_async_code:
+            return
         if not self._closed:
-            if not Util.is_async_code:
-                deprecation_warn(
-                    "Relying on Driver's destructor to close the session "
-                    "is deprecated. Please make sure to close the session. "
-                    "Use it as a context (`with` statement) or make sure to "
-                    "call `.close()` explicitly. Future versions of the "
-                    "driver will not close drivers automatically."
-                )
-                self.close()
+            _deprecation_warn(
+                "Relying on Driver's destructor to close the session "
+                "is deprecated. Please make sure to close the session. "
+                "Use it as a context (`with` statement) or make sure to "
+                "call `.close()` explicitly. Future versions of the "
+                "driver will not close drivers automatically."
+            )
+            self.close()
 
     def _check_state(self):
         if self._closed:

src/neo4j/_sync/work/workspace.py

@@ -59,15 +59,19 @@ def __init__(self, pool, config):
         self._closed = False
         super().__init__()
 
-    def __del__(self):
+    def __del__(
+        self, _unclosed_resource_warn=unclosed_resource_warn,
+        _is_async_code=Util.is_async_code,
+        _deprecation_warn=deprecation_warn,
+    ):
         if self._closed:
             return
-        unclosed_resource_warn(self)
+        _unclosed_resource_warn(self)
         # TODO: 6.0 - remove this
-        if asyncio.iscoroutinefunction(self.close):
+        if _is_async_code:
             return
         try:
-            deprecation_warn(
+            _deprecation_warn(
                 "Relying on Session's destructor to close the session "
                 "is deprecated. Please make sure to close the session. Use it "
                 "as a context (`with` statement) or make sure to call "