fix(config): add cache_config field to SQLAlchemy configs (#731)

Closes #730.

## Summary

- `SQLAlchemyAsyncConfig` and `SQLAlchemySyncConfig` now accept
`cache_config: Optional[CacheConfig]` and `cache_manager:
Optional[CacheManager]` kwargs, closing the documentation/implementation
gap from #636.
- When `cache_config` is set, `__post_init__` instantiates a
`CacheManager` (unless one was passed explicitly) and publishes it to
sessions via `session_config.info["cache_manager"]`. Repositories pick
it up automatically through the existing
`session.info.get("cache_manager")` lookup in `repository/_async.py` and
`repository/_sync.py`.
- `session_config` and its `info` dict are 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__` now includes `id(cache_manager)` so configs that differ
only by cache region remain distinguishable in the class-level
engine/session registries.
- `CacheConfig` / `CacheManager` stay behind `TYPE_CHECKING` at module
load and are imported at runtime only when `cache_config` is set, so
dogpile.cache probing is still avoided for users who don't opt in.

After this change the "Quick Start" and "Repository Integration"
examples in `docs/usage/caching.rst` work as documented:

```python
db_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///app.db",
    cache_config=CacheConfig(
        backend="dogpile.cache.memory",
        expiration_time=300,
    ),
)
async with db_config.get_session() as session:
    repo = UserRepository(session=session, auto_expunge=True)
    user = await repo.get(user_id)  # DB + cache populate
    user = await repo.get(user_id)  # cache hit (fresh deserialized instance)
```

> [!NOTE]
> `auto_expunge=True` on the repository is required for cache hits to
round-trip cleanly across consecutive calls — without it, a cached
entity attached to a closed session triggers `MissingGreenlet` on
attribute refresh in async usage. This matches the existing "Recommended
with caching" guidance in the caching guide; the Quick Start example in
the docs should likely be updated to show it inline (follow-up).

## Test plan

- [x] Repro from the issue body runs without raising on both async and
sync configs.
- [x] New unit tests in
`tests/unit/test_config/test_{async,sync}_config.py`:
- `cache_config` builds a manager and propagates it to
`session_config.info`.
- An explicit `cache_manager` overrides `cache_config` and is preserved
by identity.
  - Without `cache_config`, no `cache_manager` key is published.
- Two configs sharing a `session_config` do not clobber each other's
`cache_manager` / `file_object_raise_on_error` and keep user-supplied
`info` keys.
- `__hash__` distinguishes configs that differ only by `cache_manager`.
- [x] New integration tests in
`tests/integration/test_cache_repository.py` covering both sync and
async paths:
- `config.get_session()` + repository, populate cache, delete the
underlying DB row, second `repo.get()` in a fresh session succeeds —
proving the served value comes from the cache region rather than the
database.
- Documented Quick Start flow: two consecutive `repo.get()` calls in one
session, second call is served from cache (fresh deserialized instance,
identity differs).
- [x] Existing cache/config/routing suites: 253 passed, 1 skipped
(pre-existing `dogpile.cache`-absent gate).

<!-- docs-preview -->

<hr>
📚 Documentation preview: <a
href="https://litestar-org.github.io/advanced-alchemy-docs-preview/731">https://litestar-org.github.io/advanced-alchemy-docs-preview/731</a>

---------

Co-authored-by: Cody Fincher <cody@litestar.dev>

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,12 +224,24 @@ 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)
-        if self.session_config.info is Empty:
-            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
+        # Detach session_config and normalize info to a private dict so config writes
+        # don't bleed between configs that share the same session_config object.
+        self.session_config = copy.copy(self.session_config)
+        configured_info = self.session_config.info
+        session_info: dict[str, Any] = (
+            {} if configured_info is Empty or configured_info is None else dict(configured_info)
+        )
+        session_info["file_object_raise_on_error"] = self.file_object_raise_on_error
+        self.session_config.info = session_info
+
+        # 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:
+            session_info["cache_manager"] = self.cache_manager
 
     def __hash__(self) -> int:  # pragma: no cover
         return hash(
@@ -217,6 +250,7 @@ def __hash__(self) -> int:  # pragma: no cover
                 self.connection_string,
                 self.engine_config.__class__.__qualname__,
                 self.bind_key,
+                id(self.cache_manager) if self.cache_manager is not None else None,
             )
         )
 

docs/changelog.rst

@@ -41,6 +41,17 @@
         and cache managers. The older ``list()`` and ``list_and_count()`` names
         remain available as deprecation wrappers until 2.0.
 
+    .. change:: configure repository caching from SQLAlchemy configs
+        :type: feature
+        :pr: 731
+        :issue: 730
+
+        Adds ``cache_config`` and ``cache_manager`` support to SQLAlchemy config
+        objects. Configured cache managers are stored in ``session.info`` so
+        repositories created from config-managed sessions pick up caching
+        consistently, including when ``session_config.info`` is explicitly
+        ``None``.
+
     .. change:: migrate object storage tests to rustfs
         :type: misc
         :pr: 732

tests/integration/test_cache_repository.py

@@ -553,3 +553,202 @@ 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.asyncio
+@pytest.mark.aiosqlite
+@pytest.mark.skipif(not DOGPILE_CACHE_INSTALLED, reason="dogpile.cache not installed")
+async def test_async_config_cache_config_same_session_double_get(
+    aiosqlite_engine: AsyncEngine,
+    request: pytest.FixtureRequest,
+) -> None:
+    """Documented example: two consecutive get() calls in one session — second is a cache hit.
+
+    Cache hit serves a freshly deserialized instance, so identity differs from the first call.
+    """
+    from advanced_alchemy.config import SQLAlchemyAsyncConfig
+
+    worker_id = get_worker_id(request)
+    CachedAuthor = get_cached_author_model("aiosqlite_same_session", worker_id)
+
+    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="same:"),
+        )
+
+        class CachedAuthorRepository(SQLAlchemyAsyncRepository[Any]):
+            model_type = CachedAuthor
+
+        async with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            author = CachedAuthor(name="Same Session Async")
+            await repo.add(author)
+            await session.commit()
+            author_id = author.id
+
+        async with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            first = await repo.get(author_id)
+            second = await repo.get(author_id)
+            assert first.name == second.name == "Same Session Async"
+            # Cache hit returns a freshly deserialized instance, not the same Python object.
+            assert first is not second
+    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)
+
+
+@pytest.mark.sqlite
+@pytest.mark.skipif(not DOGPILE_CACHE_INSTALLED, reason="dogpile.cache not installed")
+def test_sync_config_cache_config_same_session_double_get(
+    sqlite_engine: Engine,
+    request: pytest.FixtureRequest,
+) -> None:
+    """Documented example: two consecutive get() calls in one session — second is a cache hit."""
+    from advanced_alchemy.config import SQLAlchemySyncConfig
+
+    worker_id = get_worker_id(request)
+    CachedAuthor = get_cached_author_model("sqlite_same_session", worker_id)
+
+    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="same:"),
+        )
+
+        class CachedAuthorRepository(SQLAlchemySyncRepository[Any]):
+            model_type = CachedAuthor
+
+        with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            author = CachedAuthor(name="Same Session Sync")
+            repo.add(author)
+            session.commit()
+            author_id = author.id
+
+        with config.get_session() as session:
+            repo = CachedAuthorRepository(session=session, auto_expunge=True)
+            first = repo.get(author_id)
+            second = repo.get(author_id)
+            assert first.name == second.name == "Same Session Sync"
+            assert first is not second
+    finally:
+        CachedAuthor.metadata.drop_all(sqlite_engine)