Fix: silent failure on entity update due to PK type mismatch

.gitignore

@@ -167,6 +167,9 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
+# VS Code
+.vscode/
+
 # Macos
 .DS_Store
 
@@ -175,4 +178,4 @@ cython_debug/
 uv.lock
 
 local_test
-crudadmin_data/
+crudadmin_data/

crudadmin/admin_interface/model_view.py

@@ -23,7 +23,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import DeclarativeBase
 
-from ..core.db import DatabaseConfig
+from ..core.db import DatabaseConfig, convert_id_to_pk_type
 from ..event import EventType, log_admin_action
 from .helper import _get_form_fields_from_schema
 
@@ -442,22 +442,7 @@ def _convert_id_to_pk_type(
         if id_value is None:
             return None
 
-        primary_key_info = self.db_config.get_primary_key_info(self.model)
-        if not primary_key_info:
-            return id_value
-
-        pk_type = primary_key_info.get("type")
-
-        if pk_type is int:
-            return int(id_value) if isinstance(id_value, str) else id_value
-        elif pk_type is str:
-            return str(id_value)
-        elif pk_type is float:
-            return float(id_value) if isinstance(id_value, str) else id_value
-        elif pk_type is UUID:
-            return UUID(str(id_value))
-        else:
-            return str(id_value)
+        return convert_id_to_pk_type(id_value, self.db_config, self.model)
 
     def setup_routes(self) -> None:
         """
@@ -571,7 +556,7 @@ def form_create_endpoint(self, template: str) -> EndpointCallable:
             ```
         """
 
-        @log_admin_action(EventType.CREATE, model=self.model)
+        @log_admin_action(EventType.CREATE, model=self.model, db_config=self.db_config)
         async def form_create_endpoint_inner(
             request: Request,
             db: AsyncSession = Depends(self.session),
@@ -756,7 +741,7 @@ def bulk_delete_endpoint(self) -> EndpointCallable:
                 - 400: Database error during deletion
         """
 
-        @log_admin_action(EventType.DELETE, model=self.model)
+        @log_admin_action(EventType.DELETE, model=self.model, db_config=self.db_config)
         async def bulk_delete_endpoint_inner(
             request: Request,
             db: AsyncSession = Depends(self.session),
@@ -1146,7 +1131,7 @@ def form_update_endpoint(self) -> EndpointCallable:
             - Supports automatic updated_at timestamp
         """
 
-        @log_admin_action(EventType.UPDATE, model=self.model)
+        @log_admin_action(EventType.UPDATE, model=self.model, db_config=self.db_config)
         async def form_update_endpoint_inner(
             request: Request,
             db: AsyncSession = Depends(self.session),

crudadmin/core/db.py

@@ -8,8 +8,10 @@
     Optional,
     Type,
     TypeVar,
+    Union,
     cast,
 )
+from uuid import UUID
 
 from fastcrud import FastCRUD
 from pydantic import BaseModel
@@ -42,6 +44,33 @@ def get_default_db_path() -> str:
     return os.path.join(data_dir, "admin.db")
 
 
+def convert_id_to_pk_type(
+    id_value: Union[int, str, None],
+    db_config: "DatabaseConfig",
+    model: Type[DeclarativeBase],
+) -> Union[int, str, float, UUID, None]:
+    """Convert the ID value to the appropriate type based on the model's primary key type."""
+    if id_value is None:
+        return None
+
+    primary_key_info = db_config.get_primary_key_info(model)
+    if not primary_key_info:
+        return id_value
+
+    pk_type = primary_key_info.get("type")
+
+    if pk_type is int:
+        return int(id_value) if isinstance(id_value, str) else id_value
+    elif pk_type is str:
+        return str(id_value)
+    elif pk_type is float:
+        return float(id_value) if isinstance(id_value, str) else id_value
+    elif pk_type is UUID:
+        return UUID(str(id_value))
+    else:
+        return str(id_value)
+
+
 class AdminBase(DeclarativeBase):
     pass
 

crudadmin/event/decorators.py

@@ -9,6 +9,8 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import DeclarativeBase
 
+from crudadmin.core.db import DatabaseConfig, convert_id_to_pk_type
+
 from .models import EventType
 
 UTC = timezone.utc
@@ -60,7 +62,9 @@ def convert_user_to_dict(user: Any) -> Dict[str, Any]:
 
 
 def log_admin_action(
-    event_type: EventType, model: Optional[Type[DeclarativeBase]] = None
+    event_type: EventType,
+    model: Optional[Type[DeclarativeBase]] = None,
+    db_config: Optional[DatabaseConfig] = None,
 ):
     def decorator(func: Callable):
         @functools.wraps(func)
@@ -88,7 +92,13 @@ async def wrapper(
 
                     if "id" in kwargs:
                         assert crud is not None, "CRUD instance should be initialized."
-                        item = await crud.get(db=db, id=kwargs["id"])
+                        request_id = kwargs["id"]
+                        if db_config:
+                            request_id = convert_id_to_pk_type(
+                                request_id, db_config, model
+                            )
+
+                        item = await crud.get(db=db, id=request_id)
                         if item:
                             previous_state = {
                                 k: v for k, v in item.items() if not k.startswith("_")

pyproject.toml

@@ -112,7 +112,8 @@ test = [
 [dependency-groups]
 dev = [
     "crudadmin",
-    "crudadmin[standard,redis,memcached,postgres,mysql,dev]"
+    "crudadmin[standard,redis,memcached,postgres,mysql,dev]",
+    "pytest-cov"
 ]
 
 [tool.pytest.ini_options]

tests/event/test_decorators.py

@@ -1,10 +1,14 @@
+import uuid
 from datetime import datetime, timezone
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi import Request
+from sqlalchemy import UUID, Column, Integer, String
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import DeclarativeBase
 
+from crudadmin.core.db import DatabaseConfig
 from crudadmin.event.decorators import (
     compare_states,
     convert_user_to_dict,
@@ -50,6 +54,36 @@ class MockModel:
     __tablename__ = "test_model"
 
 
+class TestBase(DeclarativeBase):
+    """Base class for test models."""
+
+    pass
+
+
+class MockUUIDModel(TestBase):
+    """Mock SQLAlchemy model with UUID primary key for testing."""
+
+    __name__ = "MockUUIDModel"
+    __tablename__ = "test_uuid_model"
+    id = Column(UUID(as_uuid=True), primary_key=True, index=True, default=uuid.uuid4)
+
+
+class MockIntModel(TestBase):
+    """Mock SQLAlchemy model with integer primary key for testing."""
+
+    __name__ = "MockIntModel"
+    __tablename__ = "test_int_model"
+    id = Column(Integer, primary_key=True, index=True)
+
+
+class MockStringModel(TestBase):
+    """Mock SQLAlchemy model with string primary key for testing."""
+
+    __name__ = "MockStringModel"
+    __tablename__ = "test_string_model"
+    id = Column(String, primary_key=True, index=True)
+
+
 @pytest.fixture
 def mock_request():
     """Create a mock request for testing."""
@@ -85,6 +119,17 @@ def mock_event_integration():
     return integration
 
 
+@pytest.fixture
+def mock_db_config():
+    mock_db_config = MagicMock()
+
+    def mocked_primary_key_info_func(model):
+        return DatabaseConfig.get_primary_key_info(mock_db_config, model)
+
+    mock_db_config.get_primary_key_info.side_effect = mocked_primary_key_info_func
+    return mock_db_config
+
+
 class TestGetModelChanges:
     """Test get_model_changes function."""
 
@@ -469,6 +514,88 @@ async def test_function(request, db, admin_db, current_user, id, **kwargs):
                 id=123,
             )
 
+    @pytest.mark.asyncio
+    async def test_log_admin_action_uuid_id_conversion_with_db_config(
+        mock_request, mock_db, mock_admin_db, mock_event_integration, mock_db_config
+    ):
+        test_uuid_str = "93c025d9-5831-413c-9460-edb3a28cc729"
+        test_uuid_obj = uuid.UUID(test_uuid_str)
+
+        with patch("crudadmin.event.decorators.FastCRUD") as mock_crud_class:
+            mock_crud = AsyncMock()
+            mock_crud_class.return_value = mock_crud
+            mock_crud.get.return_value = {"id": test_uuid_obj, "name": "uuid_item"}
+
+            @log_admin_action(EventType.UPDATE, MockUUIDModel, db_config=mock_db_config)
+            async def mock_endpoint(request, db, admin_db, current_user, id, **kwargs):
+                return {"status": "success"}
+
+            await mock_endpoint(
+                request=mock_request,
+                db=mock_db,
+                admin_db=mock_admin_db,
+                current_user={"id": 1},
+                event_integration=mock_event_integration,
+                id=test_uuid_str,
+            )
+
+            mock_crud.get.assert_called_with(db=mock_db, id=test_uuid_obj)
+
+    @pytest.mark.asyncio
+    async def test_log_admin_action_int_id_conversion_with_db_config(
+        mock_request, mock_db, mock_admin_db, mock_event_integration, mock_db_config
+    ):
+        input_id_str = "123"
+        expected_id_int = 123
+
+        with patch("crudadmin.event.decorators.FastCRUD") as mock_crud_class:
+            mock_crud = AsyncMock()
+            mock_crud_class.return_value = mock_crud
+            mock_crud.get.return_value = {"id": expected_id_int, "name": "int_item"}
+
+            @log_admin_action(EventType.UPDATE, MockIntModel, db_config=mock_db_config)
+            async def mock_endpoint(request, db, admin_db, current_user, id, **kwargs):
+                return {"status": "success"}
+
+            await mock_endpoint(
+                request=mock_request,
+                db=mock_db,
+                admin_db=mock_admin_db,
+                current_user={"id": 1},
+                event_integration=mock_event_integration,
+                id=input_id_str,
+            )
+
+            mock_crud.get.assert_called_with(db=mock_db, id=expected_id_int)
+
+    @pytest.mark.asyncio
+    async def test_log_admin_action_str_id_conversion_with_db_config(
+        mock_request, mock_db, mock_admin_db, mock_event_integration, mock_db_config
+    ):
+        input_id = "slug-path-id"
+
+        with patch("crudadmin.event.decorators.FastCRUD") as mock_crud_class:
+            mock_crud = AsyncMock()
+            mock_crud_class.return_value = mock_crud
+            mock_crud.get.return_value = {"id": input_id, "name": "str_item"}
+
+            @log_admin_action(
+                EventType.UPDATE, MockStringModel, db_config=mock_db_config
+            )
+            async def mock_endpoint(request, db, admin_db, current_user, id, **kwargs):
+                return {"status": "success"}
+
+            await mock_endpoint(
+                request=mock_request,
+                db=mock_db,
+                admin_db=mock_admin_db,
+                current_user={"id": 1},
+                event_integration=mock_event_integration,
+                id=input_id,
+            )
+
+            mock_crud.get.assert_called_with(db=mock_db, id=input_id)
+
 
 class TestLogAuthActionDecorator:
     """Test log_auth_action decorator."""