Implement flash messages

sqladmin/application.py

@@ -37,6 +37,9 @@
 from sqladmin.authentication import AuthenticationBackend, login_required
 from sqladmin.forms import WTFORMS_ATTRS, WTFORMS_ATTRS_REVERSED
 from sqladmin.helpers import (
+    AlertTypeEnum,
+    flash,
+    get_flashed_messages,
     get_object_identifier,
     is_async_session_maker,
     slugify_action_name,
@@ -116,6 +119,7 @@ def init_templating_engine(self) -> Jinja2Templates:
         templates.env.globals["admin"] = self
         templates.env.globals["is_list"] = lambda x: isinstance(x, list)
         templates.env.globals["get_object_identifier"] = get_object_identifier
+        templates.env.globals['get_flashed_messages'] = get_flashed_messages
 
         return templates
 
@@ -493,6 +497,8 @@ async def delete(self, request: Request) -> Response:
 
             await model_view.delete_model(request, pk)
 
+        flash(request, f"{model_view.name} deleted", AlertTypeEnum.info)
+
         referer_url = URL(request.headers.get("referer", ""))
         referer_params = MultiDict(parse_qsl(referer_url.query))
         url = URL(str(request.url_for("admin:list", identity=identity)))
@@ -530,6 +536,7 @@ async def create(self, request: Request) -> Response:
         form_data_dict = self._denormalize_wtform_data(form.data, model_view.model)
         try:
             obj = await model_view.insert_model(request, form_data_dict)
+            flash(request, f"{model_view.name} created", AlertTypeEnum.success)
         except Exception as e:
             logger.exception(e)
             context["error"] = str(e)
@@ -582,10 +589,12 @@ async def edit(self, request: Request) -> Response:
         try:
             if model_view.save_as and form_data.get("save") == "Save as new":
                 obj = await model_view.insert_model(request, form_data_dict)
+                flash(request, f"{model_view.name} created", AlertTypeEnum.success)
             else:
                 obj = await model_view.update_model(
                     request, pk=request.path_params["pk"], data=form_data_dict
                 )
+                flash(request, f"{model_view.name} updated", AlertTypeEnum.success)
         except Exception as e:
             logger.exception(e)
             context["error"] = str(e)

sqladmin/helpers.py

@@ -17,6 +17,7 @@
     TypeVar,
 )
 
+from starlette.requests import Request
 from sqlalchemy import Column, inspect
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import RelationshipProperty, sessionmaker
@@ -26,6 +27,13 @@
 T = TypeVar("T")
 
 
+class AlertTypeEnum(enum.Enum):
+    success = "success"
+    info = "info"
+    warning = "warning"
+    danger = "danger"
+
+
 _filename_ascii_strip_re = re.compile(r"[^A-Za-z0-9_.-]")
 _windows_device_files = (
     "CON",
@@ -309,3 +317,14 @@ def choice_coerce(value: Any) -> Any:
 
 def is_async_session_maker(session_maker: sessionmaker) -> bool:
     return AsyncSession in session_maker.class_.__mro__
+
+
+def flash(request: Request, message: str, alert_type: AlertTypeEnum) -> None:
+    if "_messages" not in request.session:
+        request.session["_messages"] = []
+
+    request.session["_messages"].append({"message": message, "alert_type": alert_type.value})
+
+
+def get_flashed_messages(request: Request):
+    return request.session.pop("_messages") if "_messages" in request.session else []

sqladmin/templates/sqladmin/layout.html

@@ -41,6 +41,12 @@ <h2 class="page-title">{{ title }}</h2>
           </div>
         </div>
         {% endblock %}
+        {% for message in get_flashed_messages(request) %}
+          <div class="alert alert-important alert-{{ message.alert_type }} alert-dismissible" role="alert">
+            <div class="d-flex">{{ message.message }}</div>
+            <a class="btn-close btn-close-white" data-bs-dismiss="alert" aria-label="close"></a>
+          </div>
+        {% endfor %}
       </div>
     </div>
     <div class="page-body flex-grow-1">

tests/test_helpers.py

@@ -1,11 +1,15 @@
 from datetime import timedelta
+from starlette.requests import Request
 from typing import Any
 
 import pytest
 from sqlalchemy import Column, ForeignKey, Integer, String
 from sqlalchemy.orm import declarative_base
 
 from sqladmin.helpers import (
+    AlertTypeEnum,
+    flash,
+    get_flashed_messages,
     get_object_identifier,
     is_falsy_value,
     object_identifier_values,
@@ -117,3 +121,20 @@ def test_case(ident):
 
     test_case("Missing;1")
     test_case("Johnson;7;A;Extra")
+
+
+def test_flash_messages():
+
+    # TODO: probably need to make sure SessionMiddleware is set?
+
+    request = Request({"type": "http"})
+
+    flash(request, "message 1", AlertTypeEnum.info)
+    flash(request, "message 2", AlertTypeEnum.danger)
+
+    messages = get_flashed_messages(request)
+    assert messages == [{"message": "message 1", "alert_type": "info"},
+                        {"message": "message 2", "alert_type": "danger"}]
+
+    messages = get_flashed_messages(request)
+    assert messages == []