Emit DeprecationWarning on using the driver after closing it

src/neo4j/_async/driver.py

@@ -489,6 +489,16 @@ def __del__(self):
                 )
                 self.close()
 
+    def _check_state(self):
+        if self._closed:
+            # TODO: 6.0 - raise the error
+            # raise DriverError("Driver closed")
+            deprecation_warn(
+                "Using a driver after it has been closed is deprecated. "
+                "Future versions of the driver will raise an error.",
+                stack_level=3
+            )
+
     @property
     def encrypted(self) -> bool:
         """Indicate whether the driver was configured to use encryption."""
@@ -535,6 +545,7 @@ def session(self, **config) -> AsyncSession:
 
             :returns: new :class:`neo4j.AsyncSession` object
             """
+            self._check_state()
             session_config = self._read_session_config(config)
             return self._session(session_config)
 
@@ -558,6 +569,7 @@ def _prepare_session_config(cls, preview_check, config_kwargs):
     async def close(self) -> None:
         """ Shut down, closing any open connections in the pool.
         """
+        self._check_state()
         try:
             await self._pool.close()
         except asyncio.CancelledError:
@@ -832,6 +844,7 @@ async def example(driver: neo4j.AsyncDriver) -> neo4j.Record::
             * Added the ``auth_`` parameter.
             * Stabilized from experimental.
         """
+        self._check_state()
         invalid_kwargs = [k for k in kwargs if
                           k[-2:-1] != "_" and k[-1:] == "_"]
         if invalid_kwargs:
@@ -962,6 +975,7 @@ async def verify_connectivity(self, **config) -> None:
                 If you need information about the remote server, use
                 :meth:`get_server_info` instead.
             """
+            self._check_state()
             if config:
                 experimental_warn(
                     "All configuration key-word arguments to "
@@ -1034,6 +1048,7 @@ async def get_server_info(self, **config) -> ServerInfo:
 
             .. versionadded:: 5.0
             """
+            self._check_state()
             if config:
                 experimental_warn(
                     "All configuration key-word arguments to "
@@ -1057,6 +1072,7 @@ async def supports_multi_db(self) -> bool:
             won't throw a :exc:`ConfigurationError` when trying to use this
             driver feature.
         """
+        self._check_state()
         session_config = self._read_session_config({}, preview_check=False)
         async with self._session(session_config) as session:
             await session._connect(READ_ACCESS)
@@ -1132,6 +1148,7 @@ async def verify_authentication(
 
             .. versionadded:: 5.8
             """
+            self._check_state()
             if config:
                 experimental_warn(
                     "All configuration key-word arguments but auth to "
@@ -1173,6 +1190,7 @@ async def supports_session_auth(self) -> bool:
 
         .. versionadded:: 5.8
         """
+        self._check_state()
         session_config = self._read_session_config({}, preview_check=False)
         async with self._session(session_config) as session:
             await session._connect(READ_ACCESS)

src/neo4j/_sync/driver.py

@@ -488,6 +488,16 @@ def __del__(self):
                 )
                 self.close()
 
+    def _check_state(self):
+        if self._closed:
+            # TODO: 6.0 - raise the error
+            # raise DriverError("Driver closed")
+            deprecation_warn(
+                "Using a driver after it has been closed is deprecated. "
+                "Future versions of the driver will raise an error.",
+                stack_level=3
+            )
+
     @property
     def encrypted(self) -> bool:
         """Indicate whether the driver was configured to use encryption."""
@@ -534,6 +544,7 @@ def session(self, **config) -> Session:
 
             :returns: new :class:`neo4j.Session` object
             """
+            self._check_state()
             session_config = self._read_session_config(config)
             return self._session(session_config)
 
@@ -557,6 +568,7 @@ def _prepare_session_config(cls, preview_check, config_kwargs):
     def close(self) -> None:
         """ Shut down, closing any open connections in the pool.
         """
+        self._check_state()
         try:
             self._pool.close()
         except asyncio.CancelledError:
@@ -831,6 +843,7 @@ def example(driver: neo4j.Driver) -> neo4j.Record::
             * Added the ``auth_`` parameter.
             * Stabilized from experimental.
         """
+        self._check_state()
         invalid_kwargs = [k for k in kwargs if
                           k[-2:-1] != "_" and k[-1:] == "_"]
         if invalid_kwargs:
@@ -961,6 +974,7 @@ def verify_connectivity(self, **config) -> None:
                 If you need information about the remote server, use
                 :meth:`get_server_info` instead.
             """
+            self._check_state()
             if config:
                 experimental_warn(
                     "All configuration key-word arguments to "
@@ -1033,6 +1047,7 @@ def get_server_info(self, **config) -> ServerInfo:
 
             .. versionadded:: 5.0
             """
+            self._check_state()
             if config:
                 experimental_warn(
                     "All configuration key-word arguments to "
@@ -1056,6 +1071,7 @@ def supports_multi_db(self) -> bool:
             won't throw a :exc:`ConfigurationError` when trying to use this
             driver feature.
         """
+        self._check_state()
         session_config = self._read_session_config({}, preview_check=False)
         with self._session(session_config) as session:
             session._connect(READ_ACCESS)
@@ -1131,6 +1147,7 @@ def verify_authentication(
 
             .. versionadded:: 5.8
             """
+            self._check_state()
             if config:
                 experimental_warn(
                     "All configuration key-word arguments but auth to "
@@ -1172,6 +1189,7 @@ def supports_session_auth(self) -> bool:
 
         .. versionadded:: 5.8
         """
+        self._check_state()
         session_config = self._read_session_config({}, preview_check=False)
         with self._session(session_config) as session:
             session._connect(READ_ACCESS)

tests/unit/async_/test_driver.py

@@ -18,6 +18,7 @@
 
 from __future__ import annotations
 
+import inspect
 import ssl
 import typing as t
 
@@ -951,3 +952,34 @@ async def test_supports_session_auth(session_cls_mock) -> None:
     session_mock = session_cls_mock.return_value.__aenter__.return_value
     connection_mock = session_mock._connection
     assert res is connection_mock.supports_re_auth
+
+
+@pytest.mark.parametrize(
+    ("method_name", "args", "kwargs"),
+    (
+        ("execute_query", ("",), {}),
+        ("session", (), {}),
+        ("verify_connectivity", (), {}),
+        ("get_server_info", (), {}),
+        ("supports_multi_db", (), {}),
+        ("supports_session_auth", (), {}),
+        ("close", (), {}),
+
+    )
+)
+@mark_async_test
+async def test_using_closed_driver_is_deprecated(
+    method_name, args, kwargs, session_cls_mock
+) -> None:
+    driver = AsyncGraphDatabase.driver("bolt://localhost")
+    await driver.close()
+
+    method = getattr(driver, method_name)
+    with pytest.warns(
+        DeprecationWarning,
+        match="Using a driver after it has been closed is deprecated."
+    ):
+        if inspect.iscoroutinefunction(method):
+            await method(*args, **kwargs)
+        else:
+            method(*args, **kwargs)

tests/unit/sync/test_driver.py

@@ -18,6 +18,7 @@
 
 from __future__ import annotations
 
+import inspect
 import ssl
 import typing as t
 
@@ -950,3 +951,34 @@ def test_supports_session_auth(session_cls_mock) -> None:
     session_mock = session_cls_mock.return_value.__enter__.return_value
     connection_mock = session_mock._connection
     assert res is connection_mock.supports_re_auth
+
+
+@pytest.mark.parametrize(
+    ("method_name", "args", "kwargs"),
+    (
+        ("execute_query", ("",), {}),
+        ("session", (), {}),
+        ("verify_connectivity", (), {}),
+        ("get_server_info", (), {}),
+        ("supports_multi_db", (), {}),
+        ("supports_session_auth", (), {}),
+        ("close", (), {}),
+
+    )
+)
+@mark_sync_test
+def test_using_closed_driver_is_deprecated(
+    method_name, args, kwargs, session_cls_mock
+) -> None:
+    driver = GraphDatabase.driver("bolt://localhost")
+    driver.close()
+
+    method = getattr(driver, method_name)
+    with pytest.warns(
+        DeprecationWarning,
+        match="Using a driver after it has been closed is deprecated."
+    ):
+        if inspect.iscoroutinefunction(method):
+            method(*args, **kwargs)
+        else:
+            method(*args, **kwargs)