Merge branch 'main' of github.com:dialvarezs/listestar-nuxt-fullstack

Author: dialvarezs

app_api/app/db.py

@@ -16,22 +16,32 @@
 from app.config import Settings, settings
 
 
-def create_sqlalchemy_config(app_settings: Settings | None = None) -> SQLAlchemyAsyncConfig:
+def create_sqlalchemy_config(
+    app_settings: Settings | None = None, pool_size: int | None = None, max_overflow: int | None = None
+) -> SQLAlchemyAsyncConfig:
     """
     Create SQLAlchemy configuration with the given settings.
 
     Args:
         app_settings: Settings instance to use. If None, uses global settings.
+        pool_size: Size of the connection pool. If None, uses SQLAlchemy default.
+        max_overflow: Maximum overflow size. If None, uses SQLAlchemy default.
 
     Returns:
         SQLAlchemy async configuration instance
     """
     if app_settings is None:
         app_settings = settings
 
+    engine_config_kwargs: dict = {"echo": False}
+    if pool_size is not None:
+        engine_config_kwargs["pool_size"] = pool_size
+    if max_overflow is not None:
+        engine_config_kwargs["max_overflow"] = max_overflow
+
     return SQLAlchemyAsyncConfig(
         connection_string=app_settings.database_url.unicode_string(),
-        engine_config=EngineConfig(echo=False),
+        engine_config=EngineConfig(**engine_config_kwargs),
         session_config=AsyncSessionConfig(expire_on_commit=False),
         before_send_handler="autocommit",
         alembic_config=AlembicAsyncConfig(
@@ -40,17 +50,21 @@ def create_sqlalchemy_config(app_settings: Settings | None = None) -> SQLAlchemy
     )
 
 
-def create_sqlalchemy_plugin(app_settings: Settings | None = None) -> SQLAlchemyPlugin:
+def create_sqlalchemy_plugin(
+    app_settings: Settings | None = None, pool_size: int | None = None, max_overflow: int | None = None
+) -> SQLAlchemyPlugin:
     """
     Create SQLAlchemy plugin with the given settings.
 
     Args:
         app_settings: Settings instance to use. If None, uses global settings.
+        pool_size: Size of the connection pool. If None, uses SQLAlchemy default.
+        max_overflow: Maximum overflow size. If None, uses SQLAlchemy default.
 
     Returns:
         SQLAlchemy plugin instance (config accessible via plugin.config[0])
     """
-    config = create_sqlalchemy_config(app_settings)
+    config = create_sqlalchemy_config(app_settings, pool_size=pool_size, max_overflow=max_overflow)
     return SQLAlchemyPlugin(config=config)
 
 

app_api/app/main.py

@@ -19,13 +19,15 @@
 from app.api.accounts.auth.security import create_oauth2_auth
 from app.api.accounts.router import accounts_router
 from app.config import Settings, settings
-from app.db import create_sqlalchemy_config, create_sqlalchemy_plugin
+from app.db import create_sqlalchemy_plugin
 
 
 def create_app(
     app_settings: Settings | None = None,
     title: str = "PM API",
     enable_structlog: bool = True,
+    pool_size: int | None = None,
+    max_overflow: int | None = None,
 ) -> Litestar:
     """
     Create and configure a Litestar application instance.
@@ -38,15 +40,19 @@ def create_app(
         app_settings: Settings instance to use. If None, uses global settings.
         title: Title for the OpenAPI documentation
         enable_structlog: Whether to enable structlog logging plugin
+        pool_size: Database connection pool size. If None, uses SQLAlchemy default.
+        max_overflow: Database connection pool max overflow. If None, uses SQLAlchemy default.
 
     Returns:
         Configured Litestar application instance
     """
     if app_settings is None:
         app_settings = settings
 
-    app_sqlalchemy_plugin = create_sqlalchemy_plugin(app_settings)
-    app_sqlalchemy_config = create_sqlalchemy_config(app_settings)
+    app_sqlalchemy_plugin = create_sqlalchemy_plugin(
+        app_settings, pool_size=pool_size, max_overflow=max_overflow
+    )
+    app_sqlalchemy_config = app_sqlalchemy_plugin.config[0]
 
     openapi_config = OpenAPIConfig(
         title=title,
@@ -63,7 +69,7 @@ def create_app(
             ),
             enable_middleware_logging=app_settings.debug,
             middleware_logging_config=LoggingMiddlewareConfig(
-                response_log_fields=["status_code", "cookies", "headers"],
+                response_log_fields=("status_code", "cookies", "headers"),
             ),
         )
     )

app_api/tests/accounts/conftest.py

@@ -10,6 +10,7 @@
 import pytest_asyncio
 from litestar.testing.client import AsyncTestClient
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.pool import NullPool
 
 from app.api.accounts.users.services import password_hasher
 from app.models.accounts import Permission, Role, User
@@ -149,7 +150,7 @@ async def client_with_roles(
     client: AsyncTestClient, session_database: dict[str, str]
 ) -> AsyncIterator[AsyncTestClient]:
     """Create a test client with pre-populated roles."""
-    engine = create_async_engine(session_database["url"], echo=False)
+    engine = create_async_engine(session_database["url"], echo=False, poolclass=NullPool)
     try:
         await _populate_roles(engine)
         yield client
@@ -162,7 +163,7 @@ async def client_with_accounts(
     client: AsyncTestClient, session_database: dict[str, str]
 ) -> AsyncIterator[AsyncTestClient]:
     """Create a test client with pre-populated users and roles."""
-    engine = create_async_engine(session_database["url"], echo=False)
+    engine = create_async_engine(session_database["url"], echo=False, poolclass=NullPool)
     try:
         await _populate_accounts(engine)
         yield client
@@ -175,7 +176,7 @@ async def authenticated_client(
     client: AsyncTestClient, session_database: dict[str, str]
 ) -> AsyncIterator[AsyncTestClient]:
     """Create a test client with pre-populated users and roles, and authenticate a user."""
-    engine = create_async_engine(session_database["url"], echo=False)
+    engine = create_async_engine(session_database["url"], echo=False, poolclass=NullPool)
     try:
         await _populate_accounts(engine)
 
@@ -204,7 +205,7 @@ async def client_with_permissions(
 ) -> AsyncIterator[AsyncTestClient]:
     """Create a test client with pre-populated users, roles, and permissions."""
 
-    engine = create_async_engine(session_database["url"], echo=False)
+    engine = create_async_engine(session_database["url"], echo=False, poolclass=NullPool)
     try:
         await _populate_permissions(engine)
         yield authenticated_client

app_api/tests/conftest.py

@@ -6,6 +6,7 @@
 from pytest_databases.docker.postgres import PostgresService
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.pool import NullPool
 
 from app.config import Settings
 
@@ -28,51 +29,74 @@ async def session_database(
     test_db_url = f"{base_url}/{db_name}"
 
     # Connect to default postgres db to create our session test database
-    default_engine = create_async_engine(f"{base_url}/postgres", echo=False, isolation_level="AUTOCOMMIT")
+    default_engine = create_async_engine(
+        f"{base_url}/postgres", echo=False, isolation_level="AUTOCOMMIT", poolclass=NullPool
+    )
     try:
         async with default_engine.connect() as conn:
             await conn.execute(text(f"CREATE DATABASE {db_name}"))
     finally:
         await default_engine.dispose()
 
     # Create database tables
-    engine = create_async_engine(test_db_url, echo=False)
+    engine = create_async_engine(test_db_url, echo=False, poolclass=NullPool)
     try:
         async with engine.begin() as conn:
             assert sqlalchemy_config.metadata is not None
             await conn.run_sync(sqlalchemy_config.metadata.create_all)
-
-        # Yield the database info for the session
-        yield {"url": test_db_url, "db_name": db_name, "base_url": base_url}
-
     finally:
         await engine.dispose()
 
-        # Drop the session test database
-        cleanup_engine = create_async_engine(f"{base_url}/postgres", echo=False, isolation_level="AUTOCOMMIT")
-        try:
-            async with cleanup_engine.connect() as conn:
-                # Terminate connections to the test database first
-                await conn.execute(
-                    text(f"""
-                    SELECT pg_terminate_backend(pid)
-                    FROM pg_stat_activity
-                    WHERE datname = '{db_name}' AND pid <> pg_backend_pid()
-                """)
-                )
-                await conn.execute(text(f"DROP DATABASE IF EXISTS {db_name}"))
-        finally:
-            await cleanup_engine.dispose()
+    # Yield the database info for the session
+    yield {"url": test_db_url, "db_name": db_name, "base_url": base_url}
+
+    # Drop the session test database
+    cleanup_engine = create_async_engine(
+        f"{base_url}/postgres", echo=False, isolation_level="AUTOCOMMIT", poolclass=NullPool
+    )
+    try:
+        async with cleanup_engine.connect() as conn:
+            # Terminate connections to the test database first
+            await conn.execute(
+                text(f"""
+                SELECT pg_terminate_backend(pid)
+                FROM pg_stat_activity
+                WHERE datname = '{db_name}' AND pid <> pg_backend_pid()
+            """)
+            )
+            await conn.execute(text(f"DROP DATABASE IF EXISTS {db_name}"))
+    finally:
+        await cleanup_engine.dispose()
 
 
 @pytest_asyncio.fixture(scope="function")
 async def clean_database(session_database: dict[str, str]) -> AsyncIterator[None]:
-    """Clean database between tests by truncating all tables."""
+    """
+    Clean database between tests by truncating all tables.
+
+    Uses two engines to avoid connection leaks:
+    1. terminator_engine: Kills orphaned connections to test DB (connects to postgres db)
+    2. clean_engine: Performs TRUNCATE operations (connects to test db)
+    """
     from app.db import sqlalchemy_config
 
-    clean_engine = create_async_engine(session_database["url"], echo=False)
+    terminator_engine = create_async_engine(
+        f"{session_database['base_url']}/postgres",
+        echo=False,
+        isolation_level="AUTOCOMMIT",
+        poolclass=NullPool,
+    )
+    clean_engine = create_async_engine(session_database["url"], echo=False, poolclass=NullPool)
 
     try:
+        # Terminate stray connections from previous tests to prevent connection pool exhaustion
+        async with terminator_engine.connect() as conn:
+            await conn.execute(
+                text(
+                    f"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '{session_database['db_name']}' AND pid <> pg_backend_pid()"
+                )
+            )
+
         # Clean tables before the test
         async with clean_engine.begin() as conn:
             assert sqlalchemy_config.metadata is not None
@@ -83,6 +107,7 @@ async def clean_database(session_database: dict[str, str]) -> AsyncIterator[None
 
     finally:
         await clean_engine.dispose()
+        await terminator_engine.dispose()
 
 
 @pytest_asyncio.fixture(scope="function")
@@ -123,6 +148,8 @@ def settings_customise_sources(
         app_settings=test_settings,
         title="Test PM API",
         enable_structlog=False,
+        pool_size=1,
+        max_overflow=0,
     )
 
     async with AsyncTestClient(app=test_app) as test_client: