Support MappedAsDataclass (#857)

* tests: to upcoming changes now fails
* fix: insert dataclass support
* Refactored tests and cleaned _queries.py

---------

Co-authored-by: Miradil Zeynalli <miradil.zeynalli@gmail.com>

Author: Goradii

sqladmin/_queries.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import dataclasses
 from typing import TYPE_CHECKING, Any
 
 import anyio
@@ -191,8 +192,20 @@ async def _delete_async(self, pk: str, request: Request) -> None:
             await session.commit()
             await self.model_view.after_model_delete(obj, request)
 
+    def _get_model_object(self, data: dict[str, Any]) -> Any:
+        if dataclasses.is_dataclass(self.model_view.model):
+            init_fields = {
+                f.name for f in dataclasses.fields(self.model_view.model) if f.init
+            }
+            data = {k: v for k, v in data.items() if k in init_fields}
+
+        else:
+            data = {}
+
+        return self.model_view.model(**data)
+
     def _insert_sync(self, data: dict[str, Any], request: Request) -> Any:
-        obj = self.model_view.model()
+        obj = self._get_model_object(data)
 
         with self.model_view.session_maker(expire_on_commit=False) as session:
             anyio.from_thread.run(
@@ -207,7 +220,7 @@ def _insert_sync(self, data: dict[str, Any], request: Request) -> Any:
             return obj
 
     async def _insert_async(self, data: dict[str, Any], request: Request) -> Any:
-        obj = self.model_view.model()
+        obj = self._get_model_object(data)
 
         async with self.model_view.session_maker(expire_on_commit=False) as session:
             await self.model_view.on_model_change(data, obj, True, request)

tests/test_views/dataclass/__init__.py

@@ -0,0 +1,5 @@
+import pytest
+from sqlalchemy import __version__ as __sa_version__
+
+if __sa_version__.startswith("1."):
+    pytest.skip("SQLAlchemy 1.4 does not support dataclasses", allow_module_level=True)

tests/test_views/dataclass/test_view_async.py

@@ -0,0 +1,88 @@
+from typing import AsyncGenerator
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import Integer, String, func, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    MappedAsDataclass,
+    mapped_column,
+)
+from starlette.applications import Starlette
+
+from sqladmin import Admin
+from sqladmin.models import ModelView
+from tests.common import async_engine as engine
+
+pytestmark = pytest.mark.anyio
+session_maker = async_sessionmaker(
+    bind=engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+)
+
+
+class Base(MappedAsDataclass, DeclarativeBase):
+    pass
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, init=False)
+    name: Mapped[str] = mapped_column(String(length=16), init=True)
+    email: Mapped[str] = mapped_column(String, unique=True, nullable=True, init=False)
+
+
+class UserAdmin(ModelView, model=User):
+    column_list = ["name", "email"]
+    column_labels = {"name": "Name", "email": "Email"}
+
+
+app = Starlette()
+admin = Admin(app=app, engine=engine)
+admin.add_view(UserAdmin)
+
+
+@pytest.fixture(autouse=True)
+async def prepare_database() -> AsyncGenerator[None, None]:
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+    await engine.dispose()
+
+
+@pytest.fixture
+async def client() -> AsyncGenerator[AsyncClient, None]:
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as c:
+        yield c
+
+
+async def test_async_create_dataclass(client: AsyncClient) -> None:
+    await client.post("/admin/user/create", data={"name": "foo", "email": "bar"})
+    stmt = select(func.count(User.id))
+    async with session_maker() as s:
+        result = await s.execute(stmt)
+    assert result.scalar_one() == 1
+
+
+async def test_update_dataclass(client: AsyncClient) -> None:
+    async with session_maker() as session:
+        user = User(name="John Doe")
+        session.add(user)
+        await session.commit()
+
+    await client.post("/admin/user/edit/1", data={"name": "foo", "email": "bar"})
+
+    stmt = select(User)
+    async with session_maker() as s:
+        result = await s.execute(stmt)
+    user = result.scalar_one()
+    assert user.name == "foo"
+    assert user.email == "bar"

tests/test_views/dataclass/test_view_sync.py

@@ -0,0 +1,81 @@
+from typing import Generator
+
+import pytest
+from sqlalchemy import (
+    Integer,
+    String,
+    func,
+    select,
+)
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    MappedAsDataclass,
+    mapped_column,
+    sessionmaker,
+)
+from starlette.applications import Starlette
+from starlette.testclient import TestClient
+
+from sqladmin import Admin, ModelView
+from tests.common import sync_engine as engine
+
+session_maker = sessionmaker(bind=engine)
+
+
+class Base(MappedAsDataclass, DeclarativeBase):
+    pass
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, init=False)
+    name: Mapped[str] = mapped_column(String(length=16), init=True)
+    email: Mapped[str] = mapped_column(String, unique=True, nullable=True, init=False)
+
+
+class UserAdmin(ModelView, model=User):
+    column_list = ["name", "email"]
+    column_labels = {"name": "Name", "email": "Email"}
+
+
+app = Starlette()
+admin = Admin(app=app, engine=engine)
+admin.add_model_view(UserAdmin)
+
+
+@pytest.fixture(autouse=True)
+def prepare_database() -> Generator[None, None, None]:
+    Base.metadata.create_all(engine)
+    yield
+    Base.metadata.drop_all(engine)
+
+
+@pytest.fixture
+def client() -> Generator[TestClient, None, None]:
+    with TestClient(app=app, base_url="http://testserver") as c:
+        yield c
+
+
+def test_sync_create_dataclass(client: TestClient) -> None:
+    client.post("/admin/user/create", data={"name": "foo", "email": "bar"})
+    stmt = select(func.count(User.id))
+    with session_maker() as s:
+        result = s.execute(stmt)
+    assert result.scalar_one() == 1
+
+
+def test_update_dataclass(client: TestClient) -> None:
+    with session_maker() as session:
+        user = User(name="John Doe")
+        session.add(user)
+        session.commit()
+
+    client.post("/admin/user/edit/1", data={"name": "foo", "email": "bar"})
+
+    stmt = select(User)
+    with session_maker() as s:
+        user = s.execute(stmt).scalar_one()
+    assert user.name == "foo"
+    assert user.email == "bar"