Fix record type checking (#770)

Author: Dreamsorcerer

aiohttp_admin/backends/abc.py

@@ -3,7 +3,6 @@
 import sys
 import warnings
 from abc import ABC, abstractmethod
-from collections.abc import Sequence
 from datetime import date, datetime, time
 from enum import Enum
 from functools import cached_property, partial
@@ -112,13 +111,14 @@ class AbstractAdminResource(ABC):
     primary_key: str
     omit_fields: set[str]
 
-    def __init__(self) -> None:
+    def __init__(self, record_type: Optional[dict[str, TypeAlias]] = None) -> None:
         if "id" in self.fields and self.primary_key != "id":
             warnings.warn("A non-PK 'id' column is likely to break the admin.", stacklevel=2)
 
-        d = {k: self._get_input_type(v) for k, v in self.inputs.items()}
         # For runtime type checking only.
-        self._record_type = TypedDict("RecordType", d, total=False)  # type: ignore[misc]
+        if record_type is None:
+            record_type = {k: Any for k in self.inputs}
+        self._record_type = TypedDict("RecordType", record_type, total=False)  # type: ignore[misc]
 
     async def filter_by_permissions(self, request: web.Request, perm_type: str,
                                     record: Record, original: Optional[Record] = None) -> Record:
@@ -319,12 +319,6 @@ async def _delete_many(self, request: web.Request) -> web.Response:
             raise web.HTTPNotFound()
         return json_response({"data": ids})
 
-    def _get_input_type(self, inp: InputState) -> TypeAlias:
-        t = INPUT_TYPES.get(inp["type"], str)
-        validators = inp.get("props", {}).get("validate", ())
-        assert isinstance(validators, Sequence)  # noqa: S101
-        return t if any(v["name"] == "required" for v in validators) else Optional[t]
-
     @cached_property
     def routes(self) -> tuple[web.RouteDef, ...]:
         """Routes to act on this resource.

aiohttp_admin/backends/sqlalchemy.py

@@ -168,6 +168,7 @@ def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, type[Declara
         self.fields = {}
         self.inputs = {}
         self.omit_fields = set()
+        record_type = {}
         for c in table.c.values():
             if c.foreign_keys:
                 field = "ReferenceField"
@@ -203,6 +204,10 @@ def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, type[Declara
                 props["validate"] = self._get_validators(table, c)
                 self.inputs[c.name] = comp(inp, props)  # type: ignore[assignment]
                 self.inputs[c.name]["show_create"] = show
+                field_type: Any = c.type.python_type
+                if c.nullable:
+                    field_type = Optional[field_type]
+                record_type[c.name] = field_type
 
         if not isinstance(model_or_table, sa.Table):
             # Append fields to represent ORM relationships.
@@ -257,7 +262,7 @@ def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, type[Declara
             raise NotImplementedError("Composite keys not supported yet.")
         self.primary_key = pk[0]
 
-        super().__init__()
+        super().__init__(record_type)
 
     @handle_errors
     async def get_list(self, params: GetListParams) -> tuple[list[Record], int]:

tests/test_backends_sqlalchemy.py

@@ -1,7 +1,7 @@
 import json
 from collections.abc import Awaitable, Callable
 from datetime import date, datetime
-from typing import Union
+from typing import Optional, Union
 
 import pytest
 import sqlalchemy as sa
@@ -439,3 +439,56 @@ class Wrong(base):  # type: ignore[misc,valid-type]
 
     with pytest.raises(ValueError, match="not an attribute"):
         permission_for(M, filters={Wrong.id: 1})
+
+
+async def test_record_type(
+    base: DeclarativeBase, aiohttp_client: Callable[[web.Application], Awaitable[TestClient]],
+    login: _Login
+) -> None:
+    class TestModel(base):  # type: ignore[misc,valid-type]
+        __tablename__ = "test"
+        id: Mapped[int] = mapped_column(primary_key=True)
+        foo: Mapped[Optional[bool]]
+        bar: Mapped[int]
+
+    app = web.Application()
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(base.metadata.create_all)
+
+    schema: aiohttp_admin.Schema = {
+        "security": {
+            "check_credentials": check_credentials,
+            "secure": False
+        },
+        "resources": ({"model": SAResource(engine, TestModel)},)
+    }
+    app["admin"] = aiohttp_admin.setup(app, schema)
+
+    admin_client = await aiohttp_client(app)
+    assert admin_client.app
+    h = await login(admin_client)
+
+    url = app["admin"].router["test_create"].url_for()
+    p = {"data": json.dumps({"foo": True, "bar": 5})}
+    async with admin_client.post(url, params=p, headers=h) as resp:
+        assert resp.status == 200
+        assert await resp.json() == {"data": {"id": 1, "foo": True, "bar": 5}}
+    p = {"data": json.dumps({"foo": None, "bar": -1})}
+    async with admin_client.post(url, params=p, headers=h) as resp:
+        assert resp.status == 200
+        assert await resp.json() == {"data": {"id": 2, "foo": None, "bar": -1}}
+
+    p = {"data": json.dumps({"foo": 5, "bar": "foo"})}
+    async with admin_client.post(url, params=p, headers=h) as resp:
+        assert resp.status == 400
+        errors = await resp.json()
+        assert any(e["loc"] == ["foo"] and e["type"] == "bool_parsing" for e in errors)
+        assert any(e["loc"] == ["bar"] and e["type"] == "int_parsing" for e in errors)
+
+    p = {"data": json.dumps({"foo": "foo", "bar": None})}
+    async with admin_client.post(url, params=p, headers=h) as resp:
+        assert resp.status == 400
+        errors = await resp.json()
+        assert any(e["loc"] == ["foo"] and e["type"] == "bool_parsing" for e in errors)
+        assert any(e["loc"] == ["bar"] and e["type"] == "int_type" for e in errors)