fix: build null and static pools from PoolSettings, and stop recommending the sync queue pool

docs/agents.md

@@ -120,7 +120,7 @@ class Transaction(AsyncSQLAlchemyUowTransaction):
 async def main() -> None:
     manager = AsyncSessionManager(
         "postgresql+asyncpg://user:pass@localhost:5432/app",
-        poolclass="async_adapted_queue",     # the default is "null" — no pooling
+        poolclass="async_adapted_queue",     # without pool_settings the default is "null" — no pooling
     )
     uow = AsyncSQLAlchemyUnitOfWork(manager.session_maker, transaction_factory=Transaction)
 
@@ -161,10 +161,11 @@ from the submodule named beside it further down.
 
 ### `AsyncSessionManager`
 
-`AsyncSessionManager(url, echo=False, poolclass="null", session_class=None,
+`AsyncSessionManager(url, echo=False, poolclass=None, session_class=None,
 expire_on_commit=False, connect_args=None, isolation_level=None, pool_settings=None,
 use_orjson=False, metrics=None, on_engine_created=None, dispose_timeout=30.0,
-search_path=None, **kwargs)` — `**kwargs` reach `create_async_engine`. `search_path`
+search_path=None, **kwargs)` — `**kwargs` reach `create_async_engine`. `poolclass=None`
+means `pool_settings.kind` when `pool_settings` is given, otherwise `"null"`. `search_path`
 attaches a `begin` listener to the engine that runs `set_config('search_path', …, true)` —
 `SET LOCAL` — as the first statement of every transaction, autobegun ones included;
 nothing is attached when it is `None`.
@@ -286,7 +287,8 @@ and raises `ValueError` on an unknown name; `PoolRegistry.list_available()` list
 `build_engine_kwargs(echo, poolclass, isolation_level, pool_settings, connect_args,
 extra_kwargs, use_orjson=False)` assembles the `create_async_engine` keywords: it always
 sets `pool_pre_ping` (from `pool_settings.pre_ping`, else `True`), copies non-`None` pool
-settings to `pool_size` / `max_overflow` / `pool_recycle` / `pool_timeout`, drops `None`
+settings to `pool_recycle` and — only for a `QueuePool` subclass — `pool_size` /
+`max_overflow` / `pool_timeout`, which `NullPool` and `StaticPool` reject; drops `None`
 values out of `connect_args`, and merges `extra_kwargs` last.
 `configure_orjson_serialization()` returns the `json_serializer` / `json_deserializer`
 pair and raises `ImportError` without orjson.
@@ -306,7 +308,7 @@ With the `settings` extra, `sqlalchemy_foundation_kit.contrib.settings` implemen
 | Model | Fields and defaults |
 |---|---|
 | `ConnectionSettings` | `host="localhost"`, `port=5432` (1–65535), `user="postgres"`, `password: SecretStr` **required**, `database: str` **required** |
-| `PoolSettings` | `kind="async_adapted_queue"`, `size=10`, `max_overflow=20`, `pre_ping=True`, `recycle=3600`, `timeout=30.0`. `kind="static"` with `max_overflow > 0` raises |
+| `PoolSettings` | `kind="async_adapted_queue"`, `size=10`, `max_overflow=20`, `pre_ping=True`, `recycle=3600`, `timeout=30.0`. `kind="static"` with `size > 1` raises: `StaticPool` is one shared connection, for tests. A fixed-size pool is `async_adapted_queue` with `max_overflow=0`. `size` / `max_overflow` / `timeout` reach queue pools only |
 | `QuerySettings` | `echo=False`, `statement_cache_size=0`, `prepared_statement_cache_size=0`, `isolation_level=None` |
 | `BasePostgresConfig` | `connection` **required**, `pool`, `query`, `application_name: str` **required**, `db_schema=None`, `use_orjson_serialization=True`, `jit=None`, `metrics_enabled=False`. `to_dsn(driver="asyncpg", mask_password=False)`; `__repr__` prints the masked DSN. `db_schema` is a `search_path` value applied per transaction (rule 16); `jit` is a startup parameter and is sent only when set |
 | `BasePostgresMigrationsConfig` | `postgres: BasePostgresConfig`, with `env_nested_delimiter="__"` and `extra="ignore"` |
@@ -368,9 +370,11 @@ decorator.
 2. **PostgreSQL over asyncpg only.** The DSN is `postgresql+asyncpg://…`.
    `create_async_session_manager` passes an asyncpg-specific `connection_class` and
    asyncpg-specific `connect_args`; another driver rejects them.
-3. **The default `poolclass` is `"null"` — no pooling at all.** That default is for tests.
-   A service wants `"async_adapted_queue"`, either as `poolclass=` or as
-   `PoolSettings.kind`, whose default is already correct.
+3. **Without `pool_settings`, the default `poolclass` is `"null"` — no pooling at all.**
+   That default is for tests. A service wants `"async_adapted_queue"`, either as
+   `poolclass=` or as `PoolSettings.kind`, whose default is already correct; when
+   `pool_settings` is passed without `poolclass`, its `kind` is used. `"queue"` and
+   `"singleton_thread"` are sync pools and raise `ArgumentError` on the async engine.
 4. **The commit belongs to the unit of work, never to a repository.** A repository
    `flush()`es; `transaction()` commits. A `session.commit()` inside a repository ends the
    transaction under everything else that was going to share it.
@@ -576,7 +580,7 @@ SQLAlchemy's through untouched.
 | `ImportError` | an extra is missing: `orjson`, `pydantic-settings`, `prometheus-client`, the OpenTelemetry instrumentations, `dishka`, `dependency-injector`. The message names the extra to install |
 | `RuntimeError` | `"AsyncSessionManager is closed"` — a session asked for after `aclose()` |
 | `ValueError` | an isolation level `normalize_isolation_level` does not know; an unregistered pool name; a pool name registered twice without `override=True`; `max_retries < 1`; a metric prefix that is not a Prometheus identifier |
-| `pydantic.ValidationError` | a `contrib.settings` model built without `password`, `database` or `application_name`, or a `static` pool with `max_overflow > 0` |
+| `pydantic.ValidationError` | a `contrib.settings` model built without `password`, `database` or `application_name`, or a `static` pool with `size > 1` |
 | `TypeError` | a value orjson cannot serialize; `manager.get_transaction()` on 0.2.0 (see above) |
 | `sqlalchemy.exc.IllegalStateChangeError` | one session driven by two tasks at once (rule 6) |
 | `sqlalchemy.exc.InvalidRequestError` | `isolation_level` on a unit-of-work method on 0.2.0 (see above) |

docs/guide/configuration.md

@@ -172,18 +172,22 @@ ConnectionSettings(
 | Parameter | Type | Default | Description |
 |-----------|------|---------|-------------|
 | `kind` | `PoolClassStr` | `"async_adapted_queue"` | Pool implementation (`async_adapted_queue`, `static`, `null`) |
-| `size` | `int` | `10` | Number of connections in the pool (minimum 1) |
-| `max_overflow` | `int` | `20` | Additional connections when pool is exhausted (0 or more) |
+| `size` | `int` | `10` | Number of connections the pool keeps open (minimum 1); queue pools only |
+| `max_overflow` | `int` | `20` | Additional connections when pool is exhausted (0 or more); queue pools only |
 | `pre_ping` | `bool` | `True` | Check connection health before use (recommended) |
 | `recycle` | `int` | `3600` | Recycle connections after N seconds (-1 = never) |
-| `timeout` | `float` | `30.0` | Seconds to wait for connection before raising error |
+| `timeout` | `float` | `30.0` | Seconds to wait for connection before raising error; queue pools only |
 
 **Pool Types:**
 
-- **`async_adapted_queue`** (default) — Standard async queue-based pool, recommended for most use cases
-- **`static`** — Fixed-size pool with no overflow (`max_overflow` must be 0)
+- **`async_adapted_queue`** (default) — Standard async queue-based pool, recommended for most use cases. With `max_overflow=0` it is a fixed-size pool
+- **`static`** — SQLAlchemy's `StaticPool`: exactly one connection, shared by every session. Concurrent sessions on it fail (`another operation is in progress`), so it is for tests, never for throughput. `size` must be 1
 - **`null`** — No pooling, creates new connection for each request (not recommended for production)
 
+`size`, `max_overflow` and `timeout` describe a queue of connections, so only the queue
+pools receive them; `static` and `null` ignore them. `recycle` and `pre_ping` apply to
+every pool.
+
 **Example:**
 
 ```python
@@ -205,14 +209,10 @@ PoolSettings(
     timeout=30.0,
 )
 
-# High-throughput: static pool (no overflow)
+# Fixed size: exactly 50 connections, never more
 PoolSettings(
-    kind="static",
     size=50,
-    max_overflow=0,  # Required for static pool
-    pre_ping=True,
-    recycle=1800,
-    timeout=45.0,
+    max_overflow=0,
 )
 ```
 
@@ -221,18 +221,16 @@ PoolSettings(
 The library validates pool settings:
 
 ```python
-# ❌ This raises ValueError
+# ❌ This raises ValueError: a static pool is one connection
 PoolSettings(
     kind="static",
-    size=10,
-    max_overflow=20,  # ❌ max_overflow must be 0 for static pool
+    size=50,
 )
 
-# ✅ Correct
+# ✅ Correct: one shared connection, for tests
 PoolSettings(
     kind="static",
-    size=10,
-    max_overflow=0,  # ✅ Valid
+    size=1,
 )
 ```
 

sqlalchemy_foundation_kit/base/engine.py

@@ -224,26 +224,25 @@ def _apply_pool_settings(
     Maps pool settings to SQLAlchemy's expected ``pool_*`` keyword arguments.
     Note: pool_pre_ping is already set in build_engine_kwargs, so not applied here.
 
-    Only applies pool size/overflow/recycle/timeout for pool classes that support them.
-    Checks if the pool class accepts these parameters via hasattr to avoid passing
-    unsupported kwargs to pools like NullPool or StaticPool.
+    ``pool_recycle`` applies to every pool class. ``pool_size``, ``max_overflow`` and
+    ``pool_timeout`` describe a queue of connections and exist only on ``QueuePool``
+    and its subclasses -- the async queue pools, a registered subclass, the class
+    ``instrument_pool_class`` builds. ``create_async_engine`` rejects them for any other
+    pool (``NullPool``, ``StaticPool``), so they are left out there.
 
     Args:
         engine_kwargs: Dictionary to update with pool configuration.
         poolclass: Pool class being configured.
         pool_settings: Pool configuration settings.
     """
-    # Build pool parameters dict
-    params = {
-        "pool_size": pool_settings.size,
-        "max_overflow": pool_settings.max_overflow,
-        "pool_recycle": pool_settings.recycle,
-        "pool_timeout": pool_settings.timeout,
-    }
+    params: dict[str, object] = {"pool_recycle": pool_settings.recycle}
+    if issubclass(poolclass, QueuePool):
+        params.update(
+            pool_size=pool_settings.size,
+            max_overflow=pool_settings.max_overflow,
+            pool_timeout=pool_settings.timeout,
+        )
 
-    # Apply only non-None parameters
-    # SQLAlchemy pool classes that don't support these params will ignore them
-    # or raise TypeError if passed, so we rely on the pool class itself to validate
     engine_kwargs.update({k: v for k, v in params.items() if v is not None})
 
 

sqlalchemy_foundation_kit/contrib/settings/postgres.py

@@ -47,8 +47,15 @@ class PoolSettings(BaseModel):
 
     Groups all connection pool-related parameters for SQLAlchemy engine.
 
+    ``size``, ``max_overflow`` and ``timeout`` apply to the queue pools
+    (``async_adapted_queue``, ``fallback_async_adapted_queue``) and are not passed to any
+    other pool. ``static`` is SQLAlchemy's ``StaticPool``: one connection shared by every
+    session, which concurrent sessions cannot use safely -- for tests, never as a
+    fixed-size pool. A fixed-size pool is ``async_adapted_queue`` with ``max_overflow=0``.
+
     Examples:
         >>> pool = PoolSettings(size=10, max_overflow=20)
+        >>> fixed = PoolSettings(size=10, max_overflow=0)
     """
 
     kind: PoolClassStr = Field(
@@ -68,8 +75,11 @@ class PoolSettings(BaseModel):
     @model_validator(mode="after")
     def _validate_pool_settings(self) -> PoolSettings:
         """Validate pool configuration constraints."""
-        if self.kind == "static" and self.max_overflow > 0:
-            raise ValueError(f"max_overflow must be 0 for static pool, got {self.max_overflow}")
+        if self.kind == "static" and self.size > 1:
+            raise ValueError(
+                f"size must be 1 for a static pool, got {self.size}: a static pool is one connection "
+                "shared by every session. For a fixed-size pool use kind='async_adapted_queue' with max_overflow=0"
+            )
         return self
 
 

sqlalchemy_foundation_kit/session/builder.py

@@ -32,7 +32,7 @@ class AsyncSessionManagerBuilder(Generic[SessionT]):
         Basic usage:
             >>> manager = (
             ...     AsyncSessionManagerBuilder("postgresql+asyncpg://...")
-            ...     .with_pool("queue")
+            ...     .with_pool("async_adapted_queue")
             ...     .with_echo(True)
             ...     .build()
             ... )
@@ -41,7 +41,7 @@ class AsyncSessionManagerBuilder(Generic[SessionT]):
             >>> manager = (
             ...     AsyncSessionManagerBuilder[CustomSession]("postgresql+asyncpg://...")
             ...     .with_session_class(CustomSession)
-            ...     .with_pool("queue", pool_settings=settings.pool)
+            ...     .with_pool("async_adapted_queue", pool_settings=settings.pool)
             ...     .with_metrics(metrics)
             ...     .with_callbacks(on_engine_created=instrument_engine)
             ...     .with_json_serialization(orjson=True)
@@ -52,7 +52,7 @@ class AsyncSessionManagerBuilder(Generic[SessionT]):
         Reusable configuration:
             >>> builder = (
             ...     AsyncSessionManagerBuilder("postgresql+asyncpg://...")
-            ...     .with_pool("queue")
+            ...     .with_pool("async_adapted_queue")
             ...     .with_metrics(metrics)
             ... )
             >>> manager1 = builder.with_echo(True).build()
@@ -100,8 +100,10 @@ def with_pool(
         """Configure connection pool.
 
         Args:
-            poolclass: Pool class name or type (e.g., "queue", "null").
-            pool_settings: Optional pool configuration settings.
+            poolclass: Pool class name or type: ``"async_adapted_queue"`` for a service,
+                ``"null"`` (the default when this is not called) for tests and scripts.
+            pool_settings: Optional pool configuration settings. ``size``,
+                ``max_overflow`` and ``timeout`` apply to the queue pools only.
 
         Returns:
             Self for method chaining.

sqlalchemy_foundation_kit/session/manager.py

@@ -226,7 +226,7 @@ def __init__(
         self,
         url: str,
         echo: bool = False,
-        poolclass: str | type = "null",
+        poolclass: str | type | None = None,
         session_class: type[SessionT] | None = None,
         expire_on_commit: bool = False,
         connect_args: dict[str, object] | None = None,
@@ -244,13 +244,18 @@ def __init__(
         Args:
             url: Database connection URL (required).
             echo: If True, SQLAlchemy will log all SQL statements (default: False).
-            poolclass: SQLAlchemy pool class or name (default: "null").
-                Use "queue" for production, "null" for testing.
+            poolclass: SQLAlchemy pool class or name. Defaults to ``pool_settings.kind``
+                when ``pool_settings`` is given, otherwise to ``"null"`` -- a new connection
+                for every checkout, which suits tests and scripts. A service wants
+                ``"async_adapted_queue"``. The sync pools (``"queue"``,
+                ``"singleton_thread"``) cannot run on an async engine.
             session_class: Custom ``AsyncSession`` subclass (default: ``AsyncSession``).
             expire_on_commit: If True, objects expire after commit (default: False).
             connect_args: Arguments passed to the database driver (default: None).
             isolation_level: Default transaction isolation level (default: None).
-            pool_settings: Pool configuration settings (default: None).
+            pool_settings: Pool configuration settings (default: None). ``size``,
+                ``max_overflow`` and ``timeout`` apply to the queue pools only; ``recycle``
+                and ``pre_ping`` apply to every pool.
             use_orjson: If True, use orjson for JSON serialization (default: False).
             metrics: Optional metrics collector (default: None). One that also implements
                 ``CheckoutWaitRecorder`` gets the pool class wrapped by
@@ -270,6 +275,8 @@ def __init__(
         self._closed = False
         self._close_lock = asyncio.Lock()
         self._dispose_timeout = dispose_timeout
+        if poolclass is None:
+            poolclass = pool_settings.kind if pool_settings is not None else "null"
         resolved_poolclass = resolve_pool_class(poolclass)
         if metrics is not None and isinstance(metrics, CheckoutWaitRecorder):
             # The wait for a connection is not observable from a pool event, so it has to

tests/integration/test_session_integration.py

@@ -578,3 +578,47 @@ async def test__create_async_session_manager__db_schema__applied_per_transaction
     assert search_path == SEARCH_PATH_SCHEMA
     assert application_name == "probe"
     assert jit == "on"
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("kind", "size"), [("async_adapted_queue", 10), ("null", 10), ("static", 1)])
+async def test__create_async_session_manager__every_pool_kind__runs_a_query(
+    postgres_container: PostgresContainer,
+    kind: str,
+    size: int,
+) -> None:
+    # Arrange: the configuration guide's pool kinds, each with a full PoolSettings
+    pytest.importorskip("pydantic_settings")
+    from pydantic import SecretStr
+
+    from sqlalchemy_foundation_kit.contrib.settings.postgres import (
+        BasePostgresConfig,
+        ConnectionSettings,
+        PoolSettings,
+    )
+    from sqlalchemy_foundation_kit.session.factories import create_async_session_manager
+
+    config = BasePostgresConfig(
+        connection=ConnectionSettings(
+            host=postgres_container.get_container_host_ip(),
+            port=int(postgres_container.get_exposed_port(postgres_container.port)),
+            user=postgres_container.username,
+            password=SecretStr(postgres_container.password),
+            database=postgres_container.dbname,
+        ),
+        application_name="probe",
+        pool=PoolSettings(kind=kind, size=size, max_overflow=0),  # type: ignore[arg-type]
+        use_orjson_serialization=False,
+    )
+
+    # Act
+    manager = create_async_session_manager(config)
+    try:
+        async with manager.get_transaction() as session:
+            result = (await session.execute(text("SELECT 1"))).scalar_one()
+    finally:
+        await manager.aclose()
+
+    # Assert
+    assert result == 1

tests/unit/base/test_engine.py

@@ -342,6 +342,53 @@ def test__build_engine_kwargs__pool_settings__applies_config() -> None:
     assert kwargs["pool_timeout"] == 30.0
 
 
+@pytest.mark.parametrize("poolclass", [NullPool, StaticPool])
+def test__build_engine_kwargs__pool_settings_for_a_pool_without_a_queue__passes_only_what_it_accepts(
+    poolclass: type,
+) -> None:
+    # Arrange
+    pool_settings = Mock(pre_ping=True, size=10, max_overflow=0, recycle=1800, timeout=30.0)
+
+    # Act
+    kwargs = build_engine_kwargs(
+        echo=False,
+        poolclass=poolclass,
+        isolation_level=None,
+        pool_settings=pool_settings,
+        connect_args=None,
+        extra_kwargs={},
+    )
+
+    # Assert: create_async_engine rejects pool_size/max_overflow/pool_timeout for these pools
+    assert "pool_size" not in kwargs
+    assert "max_overflow" not in kwargs
+    assert "pool_timeout" not in kwargs
+    assert kwargs["pool_recycle"] == 1800
+    assert kwargs["pool_pre_ping"] is True
+
+
+@pytest.mark.parametrize("poolclass", [AsyncAdaptedQueuePool, FallbackAsyncAdaptedQueuePool])
+def test__build_engine_kwargs__pool_settings_for_an_async_queue_pool__passes_every_setting(poolclass: type) -> None:
+    # Arrange
+    pool_settings = Mock(pre_ping=True, size=10, max_overflow=0, recycle=1800, timeout=30.0)
+
+    # Act
+    kwargs = build_engine_kwargs(
+        echo=False,
+        poolclass=poolclass,
+        isolation_level=None,
+        pool_settings=pool_settings,
+        connect_args=None,
+        extra_kwargs={},
+    )
+
+    # Assert
+    assert kwargs["pool_size"] == 10
+    assert kwargs["max_overflow"] == 0
+    assert kwargs["pool_timeout"] == 30.0
+    assert kwargs["pool_recycle"] == 1800
+
+
 def test__build_engine_kwargs__pool_settings_none_values__skips_none() -> None:
     # Arrange
     pool_settings = Mock()