feat(connection): add connection pool context awareness support

.github/workflows/test.yml

@@ -65,10 +65,10 @@ jobs:
           pip install -r requirements.txt
           pip install -r requirements-dev.txt
 
-      - name: Install activerecord and testsuite from PyPI
+      - name: Install activerecord and testsuite from git
         run: |
-          pip install rhosocial-activerecord==1.0.0.dev21
-          pip install rhosocial-activerecord-testsuite==1.0.0.dev8
+          pip install git+https://github.com/rhosocial/python-activerecord@release/v1.0.0.dev22
+          pip install git+https://github.com/rhosocial/python-activerecord-testsuite@release/v1.0.0.dev9
 
       - name: Create postgres scenario config
         run: |

changelog.d/19.added.md

@@ -0,0 +1 @@
+Added connection pool context awareness support for PostgreSQL backend, enabling proper connection management across sync/async contexts and transaction boundaries.

tests/providers/basic_connection.py

@@ -0,0 +1,161 @@
+# tests/providers/basic_connection.py
+"""
+Concrete implementation of IBasicConnectionProvider for PostgreSQL backend.
+
+This provider sets up connection pools and models for testing
+ActiveRecord context awareness.
+"""
+from typing import Type, Tuple, Optional, List
+
+from rhosocial.activerecord.model import ActiveRecord, AsyncActiveRecord
+from rhosocial.activerecord.backend.impl.postgres import PostgresBackend
+from rhosocial.activerecord.backend.impl.postgres.config import PostgresConnectionConfig
+from rhosocial.activerecord.connection.pool import BackendPool, AsyncBackendPool, PoolConfig
+from rhosocial.activerecord.backend.options import ExecutionOptions
+from rhosocial.activerecord.backend.schema import StatementType
+
+from rhosocial.activerecord.testsuite.feature.basic.connection.interfaces import IBasicConnectionProvider
+from .scenarios import get_scenario, get_enabled_scenarios
+
+
+class SyncTestUser(ActiveRecord):
+    """Sync test user model for connection pool tests."""
+    __table_name__ = "test_users"
+    id: Optional[int] = None
+    name: str
+    email: str
+
+
+class AsyncTestUser(AsyncActiveRecord):
+    """Async test user model for connection pool tests."""
+    __table_name__ = "test_users"
+    id: Optional[int] = None
+    name: str
+    email: str
+
+
+class BasicConnectionProvider(IBasicConnectionProvider):
+    """
+    PostgreSQL backend implementation for basic connection pool context tests.
+    """
+
+    def __init__(self):
+        self._active_backends: List = []
+        self._active_async_backends: List = []
+
+    def get_test_scenarios(self) -> list:
+        """Returns available test scenarios."""
+        return list(get_enabled_scenarios().keys())
+
+    def _create_test_table(self, backend):
+        """Create the test_users table."""
+        try:
+            backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL))
+        except Exception:
+            pass
+
+        backend.execute("""
+            CREATE TABLE test_users (
+                id SERIAL PRIMARY KEY,
+                name VARCHAR(255) NOT NULL,
+                email VARCHAR(255) NOT NULL
+            )
+        """, options=ExecutionOptions(stmt_type=StatementType.DDL))
+
+    async def _create_test_table_async(self, backend):
+        """Create the test_users table asynchronously."""
+        try:
+            await backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL))
+        except Exception:
+            pass
+
+        await backend.execute("""
+            CREATE TABLE test_users (
+                id SERIAL PRIMARY KEY,
+                name VARCHAR(255) NOT NULL,
+                email VARCHAR(255) NOT NULL
+            )
+        """, options=ExecutionOptions(stmt_type=StatementType.DDL))
+
+    def setup_sync_pool_and_model(self, scenario_name: str) -> Tuple[BackendPool, Type[ActiveRecord]]:
+        """Setup sync connection pool and model for context tests."""
+        _, config = get_scenario(scenario_name)
+
+        # Create connection pool
+        pool_config = PoolConfig(
+            min_size=1,
+            max_size=5,
+            backend_factory=lambda: PostgresBackend(connection_config=config)
+        )
+        pool = BackendPool.create(pool_config)
+
+        # Create table
+        with pool.connection() as backend:
+            self._create_test_table(backend)
+            self._active_backends.append(backend)
+
+        # Configure model to use a separate backend (not pool backend)
+        # This tests that model.backend() returns pool context backend
+        SyncTestUser.configure(config, PostgresBackend)
+        self._active_backends.append(SyncTestUser.__backend__)
+
+        return pool, SyncTestUser
+
+    async def setup_async_pool_and_model(self, scenario_name: str) -> Tuple[AsyncBackendPool, Type[AsyncActiveRecord]]:
+        """Setup async connection pool and model for context tests."""
+        from rhosocial.activerecord.backend.impl.postgres import AsyncPostgresBackend
+
+        _, config = get_scenario(scenario_name)
+
+        # Create connection pool
+        pool_config = PoolConfig(
+            min_size=1,
+            max_size=5,
+            backend_factory=lambda: AsyncPostgresBackend(connection_config=config)
+        )
+
+        # Create pool
+        pool = await AsyncBackendPool.create(pool_config)
+
+        # Create table
+        async with pool.connection() as backend:
+            await self._create_test_table_async(backend)
+            self._active_async_backends.append(backend)
+
+        # Configure model
+        await AsyncTestUser.configure(config, AsyncPostgresBackend)
+        self._active_async_backends.append(AsyncTestUser.__backend__)
+
+        return pool, AsyncTestUser
+
+    def cleanup_sync(self, scenario_name: str, pool: BackendPool):
+        """Cleanup after sync tests."""
+        pool.close(timeout=1.0)
+
+        # Disconnect backends
+        for backend in self._active_backends:
+            try:
+                backend.disconnect()
+            except Exception:
+                pass
+        self._active_backends.clear()
+
+    async def cleanup_async(self, scenario_name: str, pool: AsyncBackendPool):
+        """Cleanup after async tests."""
+        await pool.close(timeout=1.0)
+
+        # Disconnect backends
+        for backend in self._active_async_backends:
+            try:
+                await backend.disconnect()
+            except Exception:
+                pass
+        self._active_async_backends.clear()
+
+    def setup_sync_pool_for_crud(self, scenario_name: str) -> Tuple[BackendPool, Type[ActiveRecord]]:
+        """Setup sync connection pool for CRUD tests."""
+        return self.setup_sync_pool_and_model(scenario_name)
+
+    async def setup_async_pool_for_crud(self, scenario_name: str) -> Tuple[AsyncBackendPool, Type[AsyncActiveRecord]]:
+        """Setup async connection pool for CRUD tests."""
+        return await self.setup_async_pool_and_model(scenario_name)

tests/providers/query_connection.py

@@ -0,0 +1,153 @@
+# tests/providers/query_connection.py
+"""
+Concrete implementation of IQueryConnectionProvider for PostgreSQL backend.
+
+This provider sets up connection pools and models for testing
+query classes context awareness (ActiveQuery, CTEQuery, SetOperationQuery).
+"""
+from typing import Type, Tuple, Optional, List
+
+from rhosocial.activerecord.model import ActiveRecord, AsyncActiveRecord
+from rhosocial.activerecord.backend.impl.postgres import PostgresBackend
+from rhosocial.activerecord.backend.impl.postgres.config import PostgresConnectionConfig
+from rhosocial.activerecord.connection.pool import BackendPool, AsyncBackendPool, PoolConfig
+from rhosocial.activerecord.backend.options import ExecutionOptions
+from rhosocial.activerecord.backend.schema import StatementType
+
+from rhosocial.activerecord.testsuite.feature.query.connection.interfaces import IQueryConnectionProvider
+from .scenarios import get_scenario, get_enabled_scenarios
+
+
+class SyncQueryTestUser(ActiveRecord):
+    """Sync test user model for query connection pool tests."""
+    __table_name__ = "test_users"
+    id: Optional[int] = None
+    name: str
+    email: str
+
+
+class AsyncQueryTestUser(AsyncActiveRecord):
+    """Async test user model for query connection pool tests."""
+    __table_name__ = "test_users"
+    id: Optional[int] = None
+    name: str
+    email: str
+
+
+class QueryConnectionProvider(IQueryConnectionProvider):
+    """
+    PostgreSQL backend implementation for query connection pool context tests.
+    """
+
+    def __init__(self):
+        self._active_backends: List = []
+        self._active_async_backends: List = []
+
+    def get_test_scenarios(self) -> list:
+        """Returns available test scenarios."""
+        return list(get_enabled_scenarios().keys())
+
+    def _create_test_table(self, backend):
+        """Create the test_users table."""
+        try:
+            backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL))
+        except Exception:
+            pass
+
+        backend.execute("""
+            CREATE TABLE test_users (
+                id SERIAL PRIMARY KEY,
+                name VARCHAR(255) NOT NULL,
+                email VARCHAR(255) NOT NULL
+            )
+        """, options=ExecutionOptions(stmt_type=StatementType.DDL))
+
+    async def _create_test_table_async(self, backend):
+        """Create the test_users table asynchronously."""
+        try:
+            await backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL))
+        except Exception:
+            pass
+
+        await backend.execute("""
+            CREATE TABLE test_users (
+                id SERIAL PRIMARY KEY,
+                name VARCHAR(255) NOT NULL,
+                email VARCHAR(255) NOT NULL
+            )
+        """, options=ExecutionOptions(stmt_type=StatementType.DDL))
+
+    def setup_sync_pool_and_model(self, scenario_name: str) -> Tuple[BackendPool, Type[ActiveRecord]]:
+        """Setup sync connection pool and model for query context tests."""
+        _, config = get_scenario(scenario_name)
+
+        # Create connection pool
+        pool_config = PoolConfig(
+            min_size=1,
+            max_size=5,
+            backend_factory=lambda: PostgresBackend(connection_config=config)
+        )
+        pool = BackendPool.create(pool_config)
+
+        # Create table
+        with pool.connection() as backend:
+            self._create_test_table(backend)
+            self._active_backends.append(backend)
+
+        # Configure model to use a separate backend (not pool backend)
+        # This tests that model.backend() returns pool context backend
+        SyncQueryTestUser.configure(config, PostgresBackend)
+        self._active_backends.append(SyncQueryTestUser.__backend__)
+
+        return pool, SyncQueryTestUser
+
+    async def setup_async_pool_and_model(self, scenario_name: str) -> Tuple[AsyncBackendPool, Type[AsyncActiveRecord]]:
+        """Setup async connection pool and model for query context tests."""
+        from rhosocial.activerecord.backend.impl.postgres import AsyncPostgresBackend
+
+        _, config = get_scenario(scenario_name)
+
+        # Create connection pool
+        pool_config = PoolConfig(
+            min_size=1,
+            max_size=5,
+            backend_factory=lambda: AsyncPostgresBackend(connection_config=config)
+        )
+
+        # Create pool
+        pool = await AsyncBackendPool.create(pool_config)
+
+        # Create table
+        async with pool.connection() as backend:
+            await self._create_test_table_async(backend)
+            self._active_async_backends.append(backend)
+
+        # Configure model
+        await AsyncQueryTestUser.configure(config, AsyncPostgresBackend)
+        self._active_async_backends.append(AsyncQueryTestUser.__backend__)
+
+        return pool, AsyncQueryTestUser
+
+    def cleanup_sync(self, scenario_name: str, pool: BackendPool):
+        """Cleanup after sync tests."""
+        pool.close(timeout=1.0)
+
+        # Disconnect backends
+        for backend in self._active_backends:
+            try:
+                backend.disconnect()
+            except Exception:
+                pass
+        self._active_backends.clear()
+
+    async def cleanup_async(self, scenario_name: str, pool: AsyncBackendPool):
+        """Cleanup after async tests."""
+        await pool.close(timeout=1.0)
+
+        # Disconnect backends
+        for backend in self._active_async_backends:
+            try:
+                await backend.disconnect()
+            except Exception:
+                pass
+        self._active_async_backends.clear()

tests/providers/registry.py

@@ -10,6 +10,8 @@
 from .events import EventsProvider
 from .mixins import MixinsProvider
 from .query import QueryProvider
+from .basic_connection import BasicConnectionProvider
+from .query_connection import QueryConnectionProvider
 
 # Create a single, global instance of the ProviderRegistry for this backend.
 provider_registry = ProviderRegistry()
@@ -20,3 +22,7 @@
 provider_registry.register("feature.events.IEventsProvider", EventsProvider)
 provider_registry.register("feature.mixins.IMixinsProvider", MixinsProvider)
 provider_registry.register("feature.query.IQueryProvider", QueryProvider)
+
+# Register connection pool context awareness providers
+provider_registry.register("feature.basic.connection.IBasicConnectionProvider", BasicConnectionProvider)
+provider_registry.register("feature.query.connection.IQueryConnectionProvider", QueryConnectionProvider)

tests/rhosocial/activerecord_postgres_test/feature/basic/connection/conftest.py

@@ -0,0 +1,8 @@
+# tests/rhosocial/activerecord_postgres_test/feature/basic/connection/conftest.py
+"""
+Pytest configuration for basic connection pool feature tests.
+
+This file imports fixtures from the corresponding testsuite, making them
+available to the tests in this directory.
+"""
+from rhosocial.activerecord.testsuite.feature.basic.connection.conftest import *

tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_context.py

@@ -0,0 +1,14 @@
+# tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_context.py
+"""
+Basic ActiveRecord Context Test Module for PostgreSQL backend.
+
+This module imports and runs the shared tests from the testsuite package,
+ensuring PostgreSQL backend compatibility for connection pool context awareness.
+"""
+from rhosocial.activerecord.testsuite.feature.basic.connection.conftest import (
+    sync_pool_and_model,
+    async_pool_and_model,
+)
+
+# Import shared tests from testsuite package
+from rhosocial.activerecord.testsuite.feature.basic.connection.test_active_record_context import *

tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_crud.py

@@ -0,0 +1,14 @@
+# tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_crud.py
+"""
+Basic ActiveRecord CRUD Test Module for PostgreSQL backend.
+
+This module imports and runs the shared tests from the testsuite package,
+ensuring PostgreSQL backend compatibility for connection pool CRUD operations.
+"""
+from rhosocial.activerecord.testsuite.feature.basic.connection.conftest import (
+    sync_pool_for_crud,
+    async_pool_for_crud,
+)
+
+# Import shared tests from testsuite package
+from rhosocial.activerecord.testsuite.feature.basic.connection.test_active_record_crud import *