fix: error message handling and isolation in repositories (#605)

Correct error message retrieval and ensure that error message overrides are isolated for different repository instances. This improves the clarity and reliability of error messages across the application.

Author: cofin

advanced_alchemy/repository/_async.py

@@ -586,7 +586,7 @@ def _get_error_messages(
             error_messages = None
         if default_messages == Empty:
             default_messages = None
-        messages = DEFAULT_ERROR_MESSAGE_TEMPLATES
+        messages = cast("ErrorMessages", dict(DEFAULT_ERROR_MESSAGE_TEMPLATES))
         if default_messages and isinstance(default_messages, dict):
             messages.update(default_messages)
         if error_messages:

advanced_alchemy/repository/_sync.py

@@ -587,7 +587,7 @@ def _get_error_messages(
             error_messages = None
         if default_messages == Empty:
             default_messages = None
-        messages = DEFAULT_ERROR_MESSAGE_TEMPLATES
+        messages = cast("ErrorMessages", dict(DEFAULT_ERROR_MESSAGE_TEMPLATES))
         if default_messages and isinstance(default_messages, dict):
             messages.update(default_messages)
         if error_messages:

advanced_alchemy/repository/memory/_async.py

@@ -99,7 +99,9 @@ def __init__(
         self.auto_expunge = auto_expunge
         self.auto_refresh = auto_refresh
         self.auto_commit = auto_commit
-        self.error_messages = self._get_error_messages(error_messages=error_messages)
+        self.error_messages = self._get_error_messages(
+            error_messages=error_messages, default_messages=self.error_messages
+        )
         self.wrap_exceptions = wrap_exceptions
         self.order_by = order_by
         self._dialect: Dialect = create_autospec(Dialect, instance=True)
@@ -121,13 +123,14 @@ def _get_error_messages(
     ) -> Optional[ErrorMessages]:
         if error_messages == Empty:
             error_messages = None
-        default_messages = cast(
-            "Optional[ErrorMessages]",
-            default_messages if default_messages != Empty else DEFAULT_ERROR_MESSAGE_TEMPLATES,
-        )
-        if error_messages is not None and default_messages is not None:
-            default_messages.update(cast("ErrorMessages", error_messages))
-        return default_messages
+        if default_messages == Empty:
+            default_messages = None
+        messages = cast("ErrorMessages", dict(DEFAULT_ERROR_MESSAGE_TEMPLATES))
+        if default_messages:
+            messages.update(cast("ErrorMessages", default_messages))
+        if error_messages:
+            messages.update(cast("ErrorMessages", error_messages))
+        return messages
 
     @classmethod
     def __database_add__(cls, identity: Any, data: ModelT) -> ModelT:

advanced_alchemy/repository/memory/_sync.py

@@ -100,7 +100,9 @@ def __init__(
         self.auto_expunge = auto_expunge
         self.auto_refresh = auto_refresh
         self.auto_commit = auto_commit
-        self.error_messages = self._get_error_messages(error_messages=error_messages)
+        self.error_messages = self._get_error_messages(
+            error_messages=error_messages, default_messages=self.error_messages
+        )
         self.wrap_exceptions = wrap_exceptions
         self.order_by = order_by
         self._dialect: Dialect = create_autospec(Dialect, instance=True)
@@ -122,13 +124,14 @@ def _get_error_messages(
     ) -> Optional[ErrorMessages]:
         if error_messages == Empty:
             error_messages = None
-        default_messages = cast(
-            "Optional[ErrorMessages]",
-            default_messages if default_messages != Empty else DEFAULT_ERROR_MESSAGE_TEMPLATES,
-        )
-        if error_messages is not None and default_messages is not None:
-            default_messages.update(cast("ErrorMessages", error_messages))
-        return default_messages
+        if default_messages == Empty:
+            default_messages = None
+        messages = cast("ErrorMessages", dict(DEFAULT_ERROR_MESSAGE_TEMPLATES))
+        if default_messages:
+            messages.update(cast("ErrorMessages", default_messages))
+        if error_messages:
+            messages.update(cast("ErrorMessages", error_messages))
+        return messages
 
     @classmethod
     def __database_add__(cls, identity: Any, data: ModelT) -> ModelT:

pyproject.toml

@@ -522,7 +522,6 @@ pythonVersion = "3.9"
 reportMissingTypeStubs = false
 reportPrivateImportUsage = false
 reportUnknownMemberType = false
-reportUnnecessaryTypeIgnoreComments = true
 reportUnusedFunction = false
 strict = ["advanced_alchemy/**/*"]
 venv = ".venv"

tests/integration/test_repository.py

@@ -3,11 +3,12 @@
 import asyncio
 import datetime
 from collections.abc import Generator
-from typing import TYPE_CHECKING, Any, Literal, Optional, Union
+from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast
+from unittest.mock import create_autospec
 from uuid import UUID
 
 import pytest
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
 from sqlalchemy.orm import Session
 from time_machine import travel
 
@@ -18,6 +19,7 @@
     SearchFilter,
 )
 from advanced_alchemy.repository import SQLAlchemyAsyncRepository
+from advanced_alchemy.repository._util import DEFAULT_ERROR_MESSAGE_TEMPLATES
 from advanced_alchemy.repository.memory import (
     SQLAlchemyAsyncMockRepository,
     SQLAlchemySyncMockRepository,
@@ -635,6 +637,64 @@ async def test_repo_error_messages(seeded_test_session_async: "tuple[AsyncSessio
         await maybe_async(author_repo.get(non_existent_id))
 
 
+def _make_mock_session(engine: AsyncEngine) -> AsyncSession:
+    session = cast(AsyncSession, create_autospec(AsyncSession, instance=True))
+    session.bind = engine
+    session.get_bind.return_value = engine
+    return session
+
+
+@pytest.mark.mock_async
+def test_repo_error_message_overrides_are_isolated(
+    mock_async_engine: AsyncEngine, uuid_models_dba: "dict[str, type]"
+) -> None:
+    author_model = cast(type[Any], uuid_models_dba["author"])
+    default_not_found = DEFAULT_ERROR_MESSAGE_TEMPLATES.get("not_found")
+
+    class BaseRepo(SQLAlchemyAsyncRepository[Any]):
+        model_type = author_model
+
+    class RepoA(BaseRepo):
+        error_messages = {"not_found": "Author A not found"}
+
+    class RepoB(BaseRepo):
+        error_messages = {"not_found": "Author B not found"}
+
+    repo_a_first = RepoA(session=_make_mock_session(mock_async_engine))
+    repo_b = RepoB(session=_make_mock_session(mock_async_engine))
+    repo_a_second = RepoA(session=_make_mock_session(mock_async_engine))
+
+    assert repo_a_first.error_messages is not DEFAULT_ERROR_MESSAGE_TEMPLATES
+    assert repo_a_first.error_messages is not repo_b.error_messages
+    assert repo_a_first.error_messages["not_found"] == "Author A not found"
+    assert repo_b.error_messages["not_found"] == "Author B not found"
+    assert repo_a_second.error_messages["not_found"] == "Author A not found"
+    assert DEFAULT_ERROR_MESSAGE_TEMPLATES["not_found"] == default_not_found
+
+
+@pytest.mark.mock_async
+def test_mock_repo_error_message_overrides_are_isolated(
+    mock_async_engine: AsyncEngine, uuid_models_dba: "dict[str, type]"
+) -> None:
+    author_model = cast(type[Any], uuid_models_dba["author"])
+
+    class BaseMockRepo(SQLAlchemyAsyncMockRepository[Any]):
+        model_type = author_model
+
+    class RepoA(BaseMockRepo):
+        error_messages = {"not_found": "Mock Author A not found"}
+
+    class RepoB(BaseMockRepo):
+        error_messages = {"not_found": "Mock Author B not found"}
+
+    repo_a = RepoA(session=_make_mock_session(mock_async_engine))
+    repo_b = RepoB(session=_make_mock_session(mock_async_engine))
+
+    assert repo_a.error_messages is not repo_b.error_messages
+    assert repo_a.error_messages["not_found"] == "Mock Author A not found"
+    assert repo_b.error_messages["not_found"] == "Mock Author B not found"
+
+
 # Comprehensive tests for GitHub issue #535 and bug_fix.md issues
 async def test_service_pydantic_partial_update_github_535(
     seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]",

tests/unit/test_repository_memory_error_messages.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from typing import Any, cast
+from unittest.mock import create_autospec
+
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
+from sqlalchemy.orm import Session
+
+from advanced_alchemy.repository._util import DEFAULT_ERROR_MESSAGE_TEMPLATES
+from advanced_alchemy.repository.memory import (
+    SQLAlchemyAsyncMockRepository,
+    SQLAlchemySyncMockRepository,
+)
+
+
+def _make_async_session() -> AsyncSession:
+    session = cast(AsyncSession, create_autospec(AsyncSession, instance=True))
+    engine = cast(AsyncEngine, create_autospec(AsyncEngine, instance=True))
+    engine.dialect.name = "mock"
+    session.bind = engine
+    session.get_bind.return_value = engine
+    return session
+
+
+def _make_sync_session() -> Session:
+    session = cast(Session, create_autospec(Session, instance=True))
+    session.bind = cast(Any, create_autospec(object, instance=True))
+    return session
+
+
+def test_async_mock_repository_error_messages_isolated() -> None:
+    class BaseRepo(SQLAlchemyAsyncMockRepository[Any]):
+        model_type = object
+
+    class RepoA(BaseRepo):
+        error_messages = {"not_found": "Async Repo A"}
+
+    class RepoB(BaseRepo):
+        error_messages = {"not_found": "Async Repo B"}
+
+    repo_a_first = RepoA(session=_make_async_session())
+    repo_b = RepoB(session=_make_async_session())
+    repo_a_second = RepoA(session=_make_async_session())
+
+    assert repo_a_first.error_messages is not DEFAULT_ERROR_MESSAGE_TEMPLATES
+    assert repo_a_first.error_messages is not repo_b.error_messages
+    assert repo_a_first.error_messages["not_found"] == "Async Repo A"
+    assert repo_b.error_messages["not_found"] == "Async Repo B"
+    assert repo_a_second.error_messages["not_found"] == "Async Repo A"
+    assert DEFAULT_ERROR_MESSAGE_TEMPLATES["not_found"] == "The requested resource was not found"
+
+
+def test_async_mock_repository_instance_override_does_not_mutate_class() -> None:
+    class Repo(SQLAlchemyAsyncMockRepository[Any]):
+        model_type = object
+        error_messages = {"other": "default other"}
+
+    repo_custom = Repo(session=_make_async_session(), error_messages={"other": "custom other"})
+    repo_plain = Repo(session=_make_async_session())
+
+    assert repo_custom.error_messages["other"] == "custom other"
+    assert repo_plain.error_messages["other"] == "default other"
+    assert Repo.error_messages["other"] == "default other"
+
+
+def test_sync_mock_repository_error_messages_isolated() -> None:
+    class BaseRepo(SQLAlchemySyncMockRepository[Any]):
+        model_type = object
+
+    class RepoA(BaseRepo):
+        error_messages = {"not_found": "Sync Repo A"}
+
+    class RepoB(BaseRepo):
+        error_messages = {"not_found": "Sync Repo B"}
+
+    repo_a_first = RepoA(session=_make_sync_session())
+    repo_b = RepoB(session=_make_sync_session())
+    repo_a_second = RepoA(session=_make_sync_session())
+
+    assert repo_a_first.error_messages is not DEFAULT_ERROR_MESSAGE_TEMPLATES
+    assert repo_a_first.error_messages is not repo_b.error_messages
+    assert repo_a_first.error_messages["not_found"] == "Sync Repo A"
+    assert repo_b.error_messages["not_found"] == "Sync Repo B"
+    assert repo_a_second.error_messages["not_found"] == "Sync Repo A"
+
+
+def test_sync_mock_repository_instance_override_does_not_mutate_class() -> None:
+    class Repo(SQLAlchemySyncMockRepository[Any]):
+        model_type = object
+        error_messages = {"duplicate_key": "sync default"}
+
+    repo_custom = Repo(session=_make_sync_session(), error_messages={"duplicate_key": "custom sync"})
+    repo_plain = Repo(session=_make_sync_session())
+
+    assert repo_custom.error_messages["duplicate_key"] == "custom sync"
+    assert repo_plain.error_messages["duplicate_key"] == "sync default"
+    assert Repo.error_messages["duplicate_key"] == "sync default"