Merge pull request #30 from peterschutt/advanced-alchemy-ext

feat: custom ModelView for AA Audit Model types.

README.md

@@ -73,3 +73,37 @@ The `SQLAdminPlugin` accepts the following arguments:
 
 Views are not added to the admin app until the Litestar application is instantiated, so you can append views to the
 `views` list until this point.
+
+## Use with Advanced-Alchemy Audit Base Variants
+
+Advanced-Alchemy (AA) provides variants of base models that include `created_at` and `updated_at` fields which enforce
+that `tzinfo` is set on the `datetime` instance passed through to SQLAlchemy.
+
+When a model is created via the SQLAdmin UI, the `created_at` and `updated_at` fields default to the current time in UTC,
+however, the `tzinfo` property of the `datetime` is not set.
+
+`sqladmin-litestar-plugin` provides a custom `ModelView` class that ensures the `tzinfo` property is set on `datetime`
+instances when the form field represents an AA `DateTimeUTC` field.
+
+Example:
+
+```python
+from __future__ import annotations
+
+from advanced_alchemy.base import UUIDAuditBase
+from sqlalchemy import Column, String
+
+from sqladmin_litestar_plugin.ext.advanced_alchemy import AuditModelView
+
+
+class Entity(UUIDAuditBase):
+    my_column = Column(String(10))
+
+
+class EntityAdmin(AuditModelView, model=Entity): ...
+```
+
+For a full working example, see the `examples/aa_audit_base` directory in this repo.
+
+The `AuditModelView` class should also be useful for models that don't depend on one of the AA audit model bases, but
+still use `DateTimeUTC` fields.

examples/aa_audit_base/main.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from advanced_alchemy.base import UUIDAuditBase
+from advanced_alchemy.extensions.litestar import SQLAlchemyPlugin
+from litestar import Litestar
+from litestar.contrib.sqlalchemy.plugins import SQLAlchemyAsyncConfig
+from sqlalchemy import Column, String
+from sqlalchemy.ext.asyncio import create_async_engine
+
+import sqladmin_litestar_plugin
+from sqladmin_litestar_plugin.ext.advanced_alchemy import AuditModelView
+
+engine = create_async_engine("sqlite+aiosqlite:///")
+
+
+class Entity(UUIDAuditBase):
+    my_column = Column(String(10))
+
+
+class EntityAdmin(AuditModelView, model=Entity): ...
+
+
+app = Litestar(
+    plugins=(
+        SQLAlchemyPlugin(config=SQLAlchemyAsyncConfig(engine_instance=engine, create_all=True)),
+        sqladmin_litestar_plugin.SQLAdminPlugin(
+            views=[EntityAdmin], engine=engine, base_url="/admin"
+        ),
+    ),
+)

pyproject.toml

@@ -13,15 +13,24 @@ python = "^3.8"
 litestar = "*"
 sqladmin = "*"
 typing-extensions = "*"
+wtforms = "*"
 
 [tool.poetry.dev-dependencies]
+advanced-alchemy = "*"
 anyio = "*"
 codespell = "*"
 coverage = "*"
 mypy = "*"
 pytest = "*"
 pytest-mock = "*"
 ruff = "*"
+sqlalchemy = "*"
+starlette = "*"
+types-wtforms = "*"
+
+[tool.poetry.group.examples.dependencies]
+aiosqlite = "*"
+uvicorn = "*"
 
 [[tool.poetry.source]]
 name = "PyPI"

scripts/tests

@@ -4,22 +4,22 @@ set -eu
 
 fn_codespell() {
   echo "+++ Running codespell"
-  codespell src/ tests/ "$@"
+  codespell examples/ src/ tests/ "$@"
 }
 
 fn_ruff () {
   echo "+++ Running ruff"
-  ruff check src/ tests/ "$@" --fix
+  ruff check examples/ src/ tests/ "$@" --fix
 }
 
 fn_fmt () {
   echo "+++ Running ruff formatter"
-  ruff format src/ tests/ "$@"
+  ruff format examples/ src/ tests/ "$@"
 }
 
 fn_mypy () {
   echo "+++ Running mypy"
-  mypy src/ tests/
+  mypy examples/ src/ tests/
 }
 
 fn_pytest () {

src/sqladmin_litestar_plugin/ext/advanced_alchemy.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from datetime import timezone
+from typing import Any
+
+from sqladmin import ModelView
+from sqladmin.forms import ModelConverter, converts
+from wtforms import DateTimeField
+
+
+class DateTimeUTCField(DateTimeField):
+    def process_formdata(self, valuelist: list[Any]) -> None:
+        super().process_formdata(valuelist)
+
+        if self.data is None:
+            return
+
+        self.data = self.data.replace(tzinfo=timezone.utc)
+
+
+class DateTimeUTCConverter(ModelConverter):
+    # mypy: error: Untyped decorator makes function "convert_date_time_utc" untyped  [misc]
+    @converts("DateTimeUTC")  # type: ignore[misc]
+    def convert_date_time_utc(self, *, kwargs: dict[str, Any], **_: Any) -> DateTimeUTCField:  # noqa: PLR6301
+        return DateTimeUTCField(**kwargs)
+
+
+class AuditModelView(ModelView):
+    form_converter = DateTimeUTCConverter

tests/conftest.py

@@ -0,0 +1,6 @@
+import pytest
+
+
+@pytest.fixture(scope="session")
+def anyio_backend() -> object:
+    return "asyncio"

tests/ext/test_advanced_alchemy.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Dict
+
+import pytest
+from advanced_alchemy.base import UUIDAuditBase
+from sqlalchemy import Column, String
+from sqlalchemy.orm import sessionmaker
+from wtforms import Form
+
+from sqladmin_litestar_plugin.ext.advanced_alchemy import AuditModelView, DateTimeUTCField
+
+
+class DummyPostData(Dict[str, Any]):
+    def getlist(self, key: str) -> list[Any]:
+        v = self[key]
+        if isinstance(v, (list, tuple)):
+            return list(v)
+        return [v]
+
+
+def test_date_time_utc_field() -> None:
+    class F(Form):
+        f = DateTimeUTCField()
+
+    form = F(DummyPostData({"f": "2021-01-01 00:00:00"}))
+    assert form.f.data == datetime(2021, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+
+
+def test_date_time_utc_field_on_no_data() -> None:
+    class F(Form):
+        f = DateTimeUTCField()
+
+    form = F(DummyPostData({"f": []}))
+    assert form.f.data is None
+
+
+@pytest.mark.anyio
+async def test_audit_model_view() -> None:
+    class MyModel(UUIDAuditBase):
+        my_column = Column(String(10))
+
+    class MyModelView(AuditModelView, model=MyModel):
+        session_maker = sessionmaker()
+
+    view = MyModelView()
+    form = await view.scaffold_form()
+
+    assert isinstance(form()._fields["created_at"], DateTimeUTCField)
+    assert isinstance(form()._fields["updated_at"], DateTimeUTCField)

tests/test_plugin.py

@@ -12,11 +12,6 @@
     from litestar.types.asgi_types import Receive, Scope, Send
 
 
-@pytest.fixture(scope="session")
-def anyio_backend() -> object:
-    return "asyncio"
-
-
 @pytest.fixture(name="plugin")
 def plugin_fixture() -> SQLAdminPlugin:
     return SQLAdminPlugin()