Merge pull request #38 from benavlabs/fix-uuid-support

fix uuid, add test

Author: igorbenav

crudadmin/admin_interface/model_view.py

@@ -12,6 +12,7 @@
     Union,
     cast,
 )
+from uuid import UUID
 
 from fastapi import APIRouter, Depends, Request, UploadFile
 from fastapi.responses import JSONResponse, RedirectResponse, Response
@@ -39,7 +40,7 @@
 class BulkDeleteRequest(BaseModel):
     """Request model for bulk delete operations containing IDs to delete."""
 
-    ids: List[int]
+    ids: List[Union[int, str]]
 
 
 class PasswordTransformer:
@@ -372,6 +373,11 @@ def __init__(
         self.model = model
         self.model_key = model.__name__
         self.router = APIRouter()
+        self.admin_model = admin_model
+        self.admin_site = admin_site
+        self.allowed_actions = allowed_actions
+        self.event_integration = event_integration
+        self.password_transformer = password_transformer
 
         get_session: Callable[[], AsyncGenerator[AsyncSession, None]]
         if self._model_is_admin_model(model):
@@ -388,12 +394,6 @@ def __init__(
         self.delete_schema = delete_schema
         self.select_schema = select_schema
 
-        self.admin_model = admin_model
-        self.admin_site = admin_site
-        self.allowed_actions = allowed_actions
-        self.event_integration = event_integration
-        self.password_transformer = password_transformer
-
         self.user_service = (
             self.admin_site.admin_user_service if self.admin_site else None
         )
@@ -432,17 +432,32 @@ def get_url_prefix(self) -> str:
         return ""
 
     def _model_is_admin_model(self, model: Type[DeclarativeBase]) -> bool:
-        """Check if the given model is one of the admin-specific models."""
-        admin_model_names = [
-            self.db_config.AdminUser.__name__,
-            self.db_config.AdminSession.__name__,
-        ]
-        if self.db_config.AdminEventLog:
-            admin_model_names.append(self.db_config.AdminEventLog.__name__)
-        if self.db_config.AdminAuditLog:
-            admin_model_names.append(self.db_config.AdminAuditLog.__name__)
-
-        return model.__name__ in admin_model_names
+        """Check if a model is considered an admin model."""
+        return self.admin_model or self.model_key.lower() in {"adminuser", "admin_user"}
+
+    def _convert_id_to_pk_type(
+        self, id_value: Union[int, str]
+    ) -> Union[int, str, float]:
+        """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 = 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 str(id_value)
+        else:
+            return str(id_value)
 
     def setup_routes(self) -> None:
         """
@@ -1063,12 +1078,14 @@ def get_model_update_page(self, template: str) -> EndpointCallable:
 
         async def get_model_update_page_inner(
             request: Request,
-            id: int,
+            id: Union[int, str],
             db: AsyncSession = Depends(self.session),
         ) -> Response:
             """Show a form to update an existing record by `id`."""
+            converted_id = self._convert_id_to_pk_type(id)
+
             item = await self.crud.get(
-                db=db, id=id, schema_to_select=self.select_schema
+                db=db, id=converted_id, schema_to_select=self.select_schema
             )
             if not item:
                 return JSONResponse(
@@ -1118,7 +1135,7 @@ async def form_update_endpoint_inner(
                 cast(Any, self.admin_site).admin_authentication.get_current_user()
             ),
             event_integration=Depends(lambda: self.event_integration),
-            id: Optional[int] = None,
+            id: Optional[Union[int, str]] = None,
         ) -> Response:
             """Handle POST form submission to update an existing record."""
             assert self.admin_site is not None
@@ -1128,8 +1145,10 @@ async def form_update_endpoint_inner(
                     status_code=422, content={"message": "No id parameter provided"}
                 )
 
+            converted_id = self._convert_id_to_pk_type(id)
+
             item = await self.crud.get(
-                db=db, id=id, schema_to_select=self.select_schema
+                db=db, id=converted_id, schema_to_select=self.select_schema
             )
             if not item:
                 return JSONResponse(
@@ -1187,29 +1206,33 @@ async def form_update_endpoint_inner(
                                     AdminUserUpdateInternal(**transformed_data)
                                 )
                                 await self.crud.update(
-                                    db=db, id=id, object=admin_update_schema
+                                    db=db, id=converted_id, object=admin_update_schema
                                 )
                             else:
                                 if self.update_internal_schema:
                                     generic_update_schema = self.update_internal_schema(
                                         **transformed_data
                                     )
                                     await self.crud.update(
-                                        db=db, id=id, object=generic_update_schema
+                                        db=db,
+                                        id=converted_id,
+                                        object=generic_update_schema,
                                     )
                                 else:
                                     dynamic_update_schema = type(
                                         "InternalSchema", (BaseModel,), {}
                                     )(**transformed_data)
                                     await self.crud.update(
-                                        db=db, id=id, object=dynamic_update_schema
+                                        db=db,
+                                        id=converted_id,
+                                        object=dynamic_update_schema,
                                     )
 
                             await db.commit()
                         else:
                             update_schema_instance = self.update_schema(**update_data)
                             await self.crud.update(
-                                db=db, id=id, object=update_schema_instance
+                                db=db, id=converted_id, object=update_schema_instance
                             )
                             await db.commit()
 

tests/conftest.py

@@ -1,3 +1,4 @@
+import uuid
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
@@ -10,12 +11,14 @@
 from fastapi.testclient import TestClient
 from pydantic import BaseModel, ConfigDict
 from sqlalchemy import (
+    UUID,
     Boolean,
     Column,
     DateTime,
     ForeignKey,
     Integer,
     String,
+    Text,
     make_url,
 )
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
@@ -124,6 +127,79 @@ class UserUpdate(BaseModel):
     is_active: Optional[bool] = None
 
 
+class UUIDModel(Base):
+    """Test model with UUID primary key."""
+
+    __tablename__ = "uuid_test_model"
+    id = Column(UUID(as_uuid=True), primary_key=True, index=True, default=uuid.uuid4)
+    name = Column(String(255))
+    description = Column(Text)
+    created_at = Column(DateTime(timezone=False), server_default=func.now())
+
+
+class EmailQueryConfig(Base):
+    """Model replicating the original user's error case."""
+
+    __tablename__ = "email_query_configs"
+    id = Column(UUID(as_uuid=True), primary_key=True, index=True, default=uuid.uuid4)
+    template_id = Column(String(255))
+    query_name = Column(String(255))
+    query_text = Column(Text)
+    query_type = Column(String(50))
+    created_at = Column(DateTime(timezone=False), server_default=func.now())
+    updated_at = Column(
+        DateTime(timezone=False), server_default=func.now(), onupdate=func.now()
+    )
+
+
+class UUIDModelCreate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    name: str
+    description: str
+
+
+class UUIDModelRead(BaseModel):
+    id: uuid.UUID
+    name: str
+    description: str
+
+
+class UUIDModelUpdate(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+
+
+class UUIDModelUpdateInternal(UUIDModelUpdate):
+    id: uuid.UUID
+
+
+class EmailQueryConfigCreate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    template_id: str
+    query_name: str
+    query_text: str
+    query_type: str
+
+
+class EmailQueryConfigRead(BaseModel):
+    id: uuid.UUID
+    template_id: str
+    query_name: str
+    query_text: str
+    query_type: str
+
+
+class EmailQueryConfigUpdate(BaseModel):
+    template_id: Optional[str] = None
+    query_name: Optional[str] = None
+    query_text: Optional[str] = None
+    query_type: Optional[str] = None
+
+
+class EmailQueryConfigUpdateInternal(EmailQueryConfigUpdate):
+    id: uuid.UUID
+
+
 def is_docker_running() -> bool:
     try:
         DockerClient()
@@ -307,6 +383,92 @@ def user_update_schema():
     return UserUpdate
 
 
+@pytest.fixture
+def uuid_model():
+    return UUIDModel
+
+
+@pytest.fixture
+def email_query_config_model():
+    return EmailQueryConfig
+
+
+@pytest.fixture
+def uuid_model_create_schema():
+    return UUIDModelCreate
+
+
+@pytest.fixture
+def uuid_model_read_schema():
+    return UUIDModelRead
+
+
+@pytest.fixture
+def uuid_model_update_schema():
+    return UUIDModelUpdate
+
+
+@pytest.fixture
+def uuid_model_update_internal_schema():
+    return UUIDModelUpdateInternal
+
+
+@pytest.fixture
+def email_query_config_create_schema():
+    return EmailQueryConfigCreate
+
+
+@pytest.fixture
+def email_query_config_read_schema():
+    return EmailQueryConfigRead
+
+
+@pytest.fixture
+def email_query_config_update_schema():
+    return EmailQueryConfigUpdate
+
+
+@pytest.fixture
+def email_query_config_update_internal_schema():
+    return EmailQueryConfigUpdateInternal
+
+
+@pytest.fixture(scope="function")
+def uuid_test_data() -> list[dict]:
+    return [
+        {
+            "id": "93c025d9-5831-413c-9460-edb3a28cc729",
+            "name": "Test Item 1",
+            "description": "First test item with UUID",
+        },
+        {
+            "id": "550e8400-e29b-41d4-a716-446655440000",
+            "name": "Test Item 2",
+            "description": "Second test item with UUID",
+        },
+    ]
+
+
+@pytest.fixture(scope="function")
+def email_query_config_data() -> list[dict]:
+    return [
+        {
+            "id": "93c025d9-5831-413c-9460-edb3a28cc729",
+            "template_id": "template_1",
+            "query_name": "User Query",
+            "query_text": "SELECT * FROM users",
+            "query_type": "select",
+        },
+        {
+            "id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
+            "template_id": "template_2",
+            "query_name": "Admin Query",
+            "query_text": "SELECT * FROM admin_users",
+            "query_type": "select",
+        },
+    ]
+
+
 @pytest_asyncio.fixture(scope="function")
 async def db_config(admin_async_session) -> AsyncGenerator[DatabaseConfig, None]:
     """Create a DatabaseConfig instance for testing."""