fix(config): add cache_config field to SQLAlchemy configs

Closes litestar-org#730.

The dogpile.cache integration (litestar-org#636) added the `advanced_alchemy.cache`
package, listeners, and repository wiring, but never extended
`SQLAlchemyAsyncConfig` / `SQLAlchemySyncConfig` with the `cache_config`
kwarg that the caching guide promotes. Every example in
`docs/usage/caching.rst` raised `TypeError` because the field did not
exist.

This adds `cache_config: Optional[CacheConfig]` and
`cache_manager: Optional[CacheManager]` to `GenericSQLAlchemyConfig`.
`__post_init__` builds a `CacheManager` from `cache_config` when one is
not supplied explicitly, then publishes it to sessions via
`session_config.info["cache_manager"]` so repositories pick it up
through the existing `session.info.get("cache_manager")` path.

`session_config` and its `info` dict are now shallow-copied during
`__post_init__` so two configs constructed with the same
`session_config` instance no longer alias each other's cache_manager or
`file_object_raise_on_error` flag. `__hash__` also includes
`id(cache_manager)` so configs that differ only by cache region remain
distinguishable in the class-level registries.

Imports of `CacheConfig` / `CacheManager` are lazy: only typed under
TYPE_CHECKING at module load, and only imported at runtime inside
`__post_init__` when `cache_config` is set, so dogpile.cache probing
is still avoided for configs that don't opt in.

Author: hasansezertasan

advanced_alchemy/config/common.py

@@ -1,3 +1,4 @@
+import copy
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, ClassVar, Generic, Optional, Union, cast
@@ -16,6 +17,7 @@
     from sqlalchemy.orm.session import JoinTransactionMode
     from sqlalchemy.sql import TableClause
 
+    from advanced_alchemy.cache import CacheConfig, CacheManager
     from advanced_alchemy.utils.dataclass import EmptyType
 
 __all__ = (
@@ -187,6 +189,25 @@ class GenericSQLAlchemyConfig(Generic[EngineT, SessionT, SessionMakerT]):
     - ``False``: Log warnings on file operation failures, don't raise exceptions
     - ``True`` (default): Raise exceptions on file operation failures
     """
+    cache_config: "Optional[CacheConfig]" = None
+    """Optional :class:`CacheConfig <advanced_alchemy.cache.CacheConfig>` for dogpile.cache integration.
+
+    When set, a :class:`CacheManager <advanced_alchemy.cache.CacheManager>` is instantiated during
+    :meth:`__post_init__` and stored in ``session_config.info["cache_manager"]``. Repositories
+    created against sessions produced by this config will pick the manager up automatically.
+
+    Requires the optional ``dogpile.cache`` dependency (``pip install advanced-alchemy[dogpile]``);
+    without it the manager falls back to a no-op :class:`NullRegion`.
+
+    .. seealso::
+        :doc:`/usage/caching`
+    """
+    cache_manager: "Optional[CacheManager]" = None
+    """Optional pre-built :class:`CacheManager <advanced_alchemy.cache.CacheManager>` instance.
+
+    Takes precedence over :attr:`cache_config`. Useful when sharing a single cache manager across
+    multiple configs.
+    """
     _SESSION_SCOPE_KEY_REGISTRY: "ClassVar[set[str]]" = field(init=False, default=cast("set[str]", set()))
     """Internal counter for ensuring unique identification of session scope keys in the class."""
     _ENGINE_APP_STATE_KEY_REGISTRY: "ClassVar[set[str]]" = field(init=False, default=cast("set[str]", set()))
@@ -203,20 +224,34 @@ def __post_init__(self) -> None:
         else:
             metadata_registry.set(self.bind_key, self.metadata)
 
-        # Store file_object_raise_on_error in session_config.info
-        # Ensure session_config.info is a dict (convert from Empty if needed)
+        # Detach session_config and its info dict from any caller-shared instance so writes
+        # below (file_object flag, cache_manager) don't bleed between configs that happened
+        # to be constructed with the same session_config object.
+        self.session_config = copy.copy(self.session_config)
         if self.session_config.info is Empty:
             self.session_config.info = {}
+        elif isinstance(self.session_config.info, dict):
+            self.session_config.info = dict(self.session_config.info)
         if isinstance(self.session_config.info, dict):
             self.session_config.info["file_object_raise_on_error"] = self.file_object_raise_on_error
 
+        # Build a CacheManager from cache_config if one wasn't supplied explicitly,
+        # then propagate it to sessions via session_config.info["cache_manager"].
+        if self.cache_manager is None and self.cache_config is not None:
+            from advanced_alchemy.cache import CacheManager
+
+            self.cache_manager = CacheManager(self.cache_config)
+        if self.cache_manager is not None and isinstance(self.session_config.info, dict):
+            self.session_config.info["cache_manager"] = self.cache_manager
+
     def __hash__(self) -> int:  # pragma: no cover
         return hash(
             (
                 self.__class__.__qualname__,
                 self.connection_string,
                 self.engine_config.__class__.__qualname__,
                 self.bind_key,
+                id(self.cache_manager) if self.cache_manager is not None else None,
             )
         )
 

tests/integration/test_cache_repository.py

@@ -553,3 +553,115 @@ class CachedAuthorRepository(SQLAlchemySyncRepository[Any]):
 
     finally:
         CachedAuthor.metadata.drop_all(sqlite_engine)
+
+
+# Config-driven cache_config integration (issue #730)
+
+
+@pytest.mark.asyncio
+@pytest.mark.aiosqlite
+@pytest.mark.skipif(not DOGPILE_CACHE_INSTALLED, reason="dogpile.cache not installed")
+async def test_async_config_cache_config_repo_auto_pickup(
+    aiosqlite_engine: AsyncEngine,
+    request: pytest.FixtureRequest,
+) -> None:
+    """SQLAlchemyAsyncConfig(cache_config=...) wires session.info so the second get() is a cache hit."""
+    from sqlalchemy import text
+
+    from advanced_alchemy.config import SQLAlchemyAsyncConfig
+
+    worker_id = get_worker_id(request)
+    CachedAuthor = get_cached_author_model("aiosqlite_cfg", worker_id)
+    table_name = CachedAuthor.__tablename__
+
+    async with aiosqlite_engine.begin() as conn:
+        await conn.run_sync(CachedAuthor.metadata.create_all)
+
+    try:
+        config = SQLAlchemyAsyncConfig(
+            engine_instance=aiosqlite_engine,
+            cache_config=CacheConfig(backend="dogpile.cache.memory", expiration_time=300, key_prefix="cfg:"),
+        )
+
+        class CachedAuthorRepository(SQLAlchemyAsyncRepository[Any]):
+            model_type = CachedAuthor
+
+        async with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            assert repo._cache_manager is config.cache_manager
+
+            author = CachedAuthor(name="Cfg Async")
+            await repo.add(author)
+            await session.commit()
+            author_id = author.id
+
+            first = await repo.get(author_id)
+            assert first.name == "Cfg Async"
+            assert config.cache_manager is not None
+            cached_region = config.cache_manager.get_entity_sync(table_name, author_id, CachedAuthor)
+            assert cached_region is not None and cached_region.name == "Cfg Async"
+
+        # Delete the row out from under SQLAlchemy. Cache survives, so the next get() can only
+        # succeed if it's served from the cache region, not the DB.
+        async with aiosqlite_engine.begin() as conn:
+            await conn.execute(text(f"DELETE FROM {table_name}"))
+
+        async with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            from_cache = await repo.get(author_id)
+            assert from_cache.name == "Cfg Async"
+    finally:
+        async with aiosqlite_engine.begin() as conn:
+            await conn.run_sync(CachedAuthor.metadata.drop_all)
+
+
+@pytest.mark.sqlite
+@pytest.mark.skipif(not DOGPILE_CACHE_INSTALLED, reason="dogpile.cache not installed")
+def test_sync_config_cache_config_repo_auto_pickup(
+    sqlite_engine: Engine,
+    request: pytest.FixtureRequest,
+) -> None:
+    """SQLAlchemySyncConfig(cache_config=...) wires session.info so the second get() is a cache hit."""
+    from sqlalchemy import text
+
+    from advanced_alchemy.config import SQLAlchemySyncConfig
+
+    worker_id = get_worker_id(request)
+    CachedAuthor = get_cached_author_model("sqlite_cfg", worker_id)
+    table_name = CachedAuthor.__tablename__
+
+    CachedAuthor.metadata.create_all(sqlite_engine)
+
+    try:
+        config = SQLAlchemySyncConfig(
+            engine_instance=sqlite_engine,
+            cache_config=CacheConfig(backend="dogpile.cache.memory", expiration_time=300, key_prefix="cfg:"),
+        )
+
+        class CachedAuthorRepository(SQLAlchemySyncRepository[Any]):
+            model_type = CachedAuthor
+
+        with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            assert repo._cache_manager is config.cache_manager
+
+            author = CachedAuthor(name="Cfg Sync")
+            repo.add(author)
+            session.commit()
+            author_id = author.id
+
+            first = repo.get(author_id)
+            assert first.name == "Cfg Sync"
+            assert config.cache_manager is not None
+            cached_region = config.cache_manager.get_entity_sync(table_name, author_id, CachedAuthor)
+            assert cached_region is not None and cached_region.name == "Cfg Sync"
+
+        with sqlite_engine.begin() as conn:
+            conn.execute(text(f"DELETE FROM {table_name}"))
+
+        with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            from_cache = repo.get(author_id)
+            assert from_cache.name == "Cfg Sync"
+    finally:
+        CachedAuthor.metadata.drop_all(sqlite_engine)

tests/unit/test_config/test_async_config.py

@@ -140,3 +140,77 @@ def test_post_init_routing_config_fallback_to_engine_configs() -> None:
 
     config = SQLAlchemyAsyncConfig(routing_config=mock_routing)
     assert config.connection_string == "sqlite+aiosqlite:///fallback"
+
+
+def test_post_init_cache_config_builds_manager() -> None:
+    """cache_config triggers CacheManager creation and propagation to session.info."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+
+    config = SQLAlchemyAsyncConfig(
+        connection_string="sqlite+aiosqlite:///:memory:",
+        cache_config=CacheConfig(backend="dogpile.cache.memory", expiration_time=300),
+    )
+    assert isinstance(config.cache_manager, CacheManager)
+    assert config.session_config.info["cache_manager"] is config.cache_manager
+
+
+def test_post_init_explicit_cache_manager_overrides() -> None:
+    """An explicit cache_manager is preserved and propagated; cache_config is not re-instantiated."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+
+    manager = CacheManager(CacheConfig(backend="dogpile.cache.memory"))
+    config = SQLAlchemyAsyncConfig(
+        connection_string="sqlite+aiosqlite:///:memory:",
+        cache_manager=manager,
+    )
+    assert config.cache_manager is manager
+    assert config.session_config.info["cache_manager"] is manager
+
+
+def test_post_init_without_cache_config_leaves_info_clean() -> None:
+    """No cache_config means no cache_manager key on session.info."""
+    config = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///:memory:")
+    assert config.cache_manager is None
+    assert "cache_manager" not in config.session_config.info
+
+
+def test_post_init_does_not_alias_shared_session_config_info() -> None:
+    """Two configs sharing a session_config must not clobber each other's cache_manager."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+    from advanced_alchemy.config.asyncio import AsyncSessionConfig
+
+    shared = AsyncSessionConfig(info={"user_key": "preserved"})
+    mgr_a = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="a:"))
+    mgr_b = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="b:"))
+
+    cfg_a = SQLAlchemyAsyncConfig(
+        connection_string="sqlite+aiosqlite:///:memory:",
+        session_config=shared,
+        cache_manager=mgr_a,
+    )
+    cfg_b = SQLAlchemyAsyncConfig(
+        connection_string="sqlite+aiosqlite:///:memory:",
+        session_config=shared,
+        cache_manager=mgr_b,
+    )
+
+    assert cfg_a.session_config.info["cache_manager"] is mgr_a
+    assert cfg_b.session_config.info["cache_manager"] is mgr_b
+    # User-supplied keys survive the defensive copy.
+    assert cfg_a.session_config.info["user_key"] == "preserved"
+    assert cfg_b.session_config.info["user_key"] == "preserved"
+
+
+def test_hash_distinguishes_configs_by_cache_manager() -> None:
+    """Configs identical except for cache_manager hash differently."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+
+    mgr_a = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="a:"))
+    mgr_b = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="b:"))
+
+    cfg_a = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///:memory:", cache_manager=mgr_a)
+    cfg_b = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///:memory:", cache_manager=mgr_b)
+    cfg_none = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///:memory:")
+
+    assert hash(cfg_a) != hash(cfg_b)
+    assert hash(cfg_a) != hash(cfg_none)

tests/unit/test_config/test_sync_config.py

@@ -128,3 +128,76 @@ def test_post_init_routing_config_fallback_to_engine_configs() -> None:
 
     config = SQLAlchemySyncConfig(routing_config=mock_routing)
     assert config.connection_string == "sqlite:///fallback"
+
+
+def test_post_init_cache_config_builds_manager() -> None:
+    """cache_config triggers CacheManager creation and propagation to session.info."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+
+    config = SQLAlchemySyncConfig(
+        connection_string="sqlite:///:memory:",
+        cache_config=CacheConfig(backend="dogpile.cache.memory", expiration_time=300),
+    )
+    assert isinstance(config.cache_manager, CacheManager)
+    assert config.session_config.info["cache_manager"] is config.cache_manager
+
+
+def test_post_init_explicit_cache_manager_overrides() -> None:
+    """An explicit cache_manager is preserved and propagated; cache_config is not re-instantiated."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+
+    manager = CacheManager(CacheConfig(backend="dogpile.cache.memory"))
+    config = SQLAlchemySyncConfig(
+        connection_string="sqlite:///:memory:",
+        cache_manager=manager,
+    )
+    assert config.cache_manager is manager
+    assert config.session_config.info["cache_manager"] is manager
+
+
+def test_post_init_without_cache_config_leaves_info_clean() -> None:
+    """No cache_config means no cache_manager key on session.info."""
+    config = SQLAlchemySyncConfig(connection_string="sqlite:///:memory:")
+    assert config.cache_manager is None
+    assert "cache_manager" not in config.session_config.info
+
+
+def test_post_init_does_not_alias_shared_session_config_info() -> None:
+    """Two configs sharing a session_config must not clobber each other's cache_manager."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+    from advanced_alchemy.config.sync import SyncSessionConfig
+
+    shared = SyncSessionConfig(info={"user_key": "preserved"})
+    mgr_a = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="a:"))
+    mgr_b = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="b:"))
+
+    cfg_a = SQLAlchemySyncConfig(
+        connection_string="sqlite:///:memory:",
+        session_config=shared,
+        cache_manager=mgr_a,
+    )
+    cfg_b = SQLAlchemySyncConfig(
+        connection_string="sqlite:///:memory:",
+        session_config=shared,
+        cache_manager=mgr_b,
+    )
+
+    assert cfg_a.session_config.info["cache_manager"] is mgr_a
+    assert cfg_b.session_config.info["cache_manager"] is mgr_b
+    assert cfg_a.session_config.info["user_key"] == "preserved"
+    assert cfg_b.session_config.info["user_key"] == "preserved"
+
+
+def test_hash_distinguishes_configs_by_cache_manager() -> None:
+    """Configs identical except for cache_manager hash differently."""
+    from advanced_alchemy.cache import CacheConfig, CacheManager
+
+    mgr_a = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="a:"))
+    mgr_b = CacheManager(CacheConfig(backend="dogpile.cache.memory", key_prefix="b:"))
+
+    cfg_a = SQLAlchemySyncConfig(connection_string="sqlite:///:memory:", cache_manager=mgr_a)
+    cfg_b = SQLAlchemySyncConfig(connection_string="sqlite:///:memory:", cache_manager=mgr_b)
+    cfg_none = SQLAlchemySyncConfig(connection_string="sqlite:///:memory:")
+
+    assert hash(cfg_a) != hash(cfg_b)
+    assert hash(cfg_a) != hash(cfg_none)