fix: add pytest-timeout dependency and update related configurations

Author: cofin

pyproject.toml

@@ -122,6 +122,7 @@ test = [
   "pytest-mock>=3.14.0",
   "pytest-sugar>=1.0.0",
   "pytest-xdist>=3.6.1",
+  "pytest-timeout>=2.3.1",
 ]
 
 [project.scripts]
@@ -280,6 +281,8 @@ exclude_lines = [
 addopts = ["-q", "-ra"]
 asyncio_default_fixture_loop_scope = "function"
 asyncio_mode = "auto"
+timeout = 300
+timeout_method = "thread"
 filterwarnings = [
   "ignore::DeprecationWarning:pkg_resources.*",
   "ignore:pkg_resources is deprecated as an API:DeprecationWarning",

sqlspec/adapters/psycopg/config.py

@@ -207,8 +207,6 @@ def _close_pool(self) -> None:
         logger.info("Closing Psycopg connection pool", extra={"adapter": "psycopg"})
 
         try:
-            self.pool_instance._closed = True  # pyright: ignore[reportPrivateUsage]
-
             self.pool_instance.close()
             logger.info("Psycopg connection pool closed successfully", extra={"adapter": "psycopg"})
         except Exception as e:
@@ -402,8 +400,6 @@ async def _close_pool(self) -> None:
             return
 
         try:
-            self.pool_instance._closed = True  # pyright: ignore[reportPrivateUsage]
-
             await self.pool_instance.close()
         finally:
             self.pool_instance = None

tests/integration/test_adapters/test_psycopg/conftest.py

@@ -8,7 +8,7 @@
 from sqlspec.adapters.psycopg import PsycopgAsyncConfig, PsycopgSyncConfig
 
 if TYPE_CHECKING:
-    from collections.abc import Generator
+    from collections.abc import AsyncGenerator, Generator
 
 
 @pytest.fixture
@@ -25,26 +25,16 @@ def psycopg_sync_config(postgres_service: PostgresService) -> "Generator[Psycopg
         config.close_pool()
 
 
-@pytest.fixture
-def psycopg_async_config(postgres_service: PostgresService) -> "Generator[PsycopgAsyncConfig, None, None]":
+@pytest.fixture(scope="function")
+async def psycopg_async_config(postgres_service: PostgresService) -> "AsyncGenerator[PsycopgAsyncConfig, None]":
     """Create a psycopg async configuration."""
     config = PsycopgAsyncConfig(
         pool_config={
             "conninfo": f"postgresql://{postgres_service.user}:{postgres_service.password}@{postgres_service.host}:{postgres_service.port}/{postgres_service.database}"
         }
     )
-    yield config
-
-    if config.pool_instance:
-        import asyncio
-
-        try:
-            loop = asyncio.get_running_loop()
-            if not loop.is_closed():
-                loop.run_until_complete(config.close_pool())
-        except RuntimeError:
-            new_loop = asyncio.new_event_loop()
-            try:
-                new_loop.run_until_complete(config.close_pool())
-            finally:
-                new_loop.close()
+    try:
+        yield config
+    finally:
+        if config.pool_instance:
+            await config.close_pool()

tests/integration/test_adapters/test_psycopg/test_migrations.py

@@ -163,17 +163,7 @@ def down():
                 assert len(result.data) == 0
         finally:
             if config.pool_instance:
-                import asyncio
-
-                try:
-                    asyncio.get_running_loop()
-                    import concurrent.futures
-
-                    with concurrent.futures.ThreadPoolExecutor() as executor:
-                        future = executor.submit(asyncio.run, config.close_pool())
-                        future.result()
-                except RuntimeError:
-                    asyncio.run(config.close_pool())
+                await config.close_pool()
 
 
 def test_psycopg_sync_multiple_migrations_workflow(postgres_service: PostgresService) -> None:
@@ -380,17 +370,7 @@ def down():
                 assert len(users_result.data) == 0
         finally:
             if config.pool_instance:
-                import asyncio
-
-                try:
-                    asyncio.get_running_loop()
-                    import concurrent.futures
-
-                    with concurrent.futures.ThreadPoolExecutor() as executor:
-                        future = executor.submit(asyncio.run, config.close_pool())
-                        future.result()
-                except RuntimeError:
-                    asyncio.run(config.close_pool())
+                await config.close_pool()
 
 
 def test_psycopg_sync_migration_current_command(postgres_service: PostgresService) -> None:
@@ -509,17 +489,7 @@ def down():
             assert current_version is None or current_version == "base"
         finally:
             if config.pool_instance:
-                import asyncio
-
-                try:
-                    asyncio.get_running_loop()
-                    import concurrent.futures
-
-                    with concurrent.futures.ThreadPoolExecutor() as executor:
-                        future = executor.submit(asyncio.run, config.close_pool())
-                        future.result()
-                except RuntimeError:
-                    asyncio.run(config.close_pool())
+                await config.close_pool()
 
 
 def test_psycopg_sync_migration_error_handling(postgres_service: PostgresService) -> None:
@@ -618,17 +588,7 @@ def down():
                     assert "no such" in str(e).lower() or "does not exist" in str(e).lower()
         finally:
             if config.pool_instance:
-                import asyncio
-
-                try:
-                    asyncio.get_running_loop()
-                    import concurrent.futures
-
-                    with concurrent.futures.ThreadPoolExecutor() as executor:
-                        future = executor.submit(asyncio.run, config.close_pool())
-                        future.result()
-                except RuntimeError:
-                    asyncio.run(config.close_pool())
+                await config.close_pool()
 
 
 def test_psycopg_sync_migration_with_transactions(postgres_service: PostgresService) -> None:
@@ -793,14 +753,4 @@ def down():
                 assert len(result.data) == 0
         finally:
             if config.pool_instance:
-                import asyncio
-
-                try:
-                    asyncio.get_running_loop()
-                    import concurrent.futures
-
-                    with concurrent.futures.ThreadPoolExecutor() as executor:
-                        future = executor.submit(asyncio.run, config.close_pool())
-                        future.result()
-                except RuntimeError:
-                    asyncio.run(config.close_pool())
+                await config.close_pool()