Add ability to specify custom actions (#486)

Co-authored-by: Amin Alaee <[email protected]>

docs/api_reference/application.md

@@ -12,3 +12,6 @@
         - add_view
         - add_model_view
         - add_base_view
+
+::: sqladmin.application.action
+    handler: python

docs/configurations.md

@@ -279,3 +279,38 @@ By default these methods do nothing.
             # Perform some other action
             ...
     ```
+
+## Custom Action
+
+To add custom action on models to the Admin interface, you can use the `action` annotation.
+
+For example:
+
+!!! example
+
+    ```python
+    from sqladmin import BaseView, action
+
+    class UserAdmin(ModelView, model=User):
+        @action(
+            name="approve_users",
+            label="Approve",
+            confirmation_message="Are you sure?",
+            add_in_detail=True,
+            add_in_list=True,
+        )
+        async def approve_users(self, request: Request):
+            pks = request.query_params.get("pks", "").split(",")
+            if pks:
+                for pk in pks:
+                    model: User = await self.get_object_for_edit(pk)
+                    ...
+
+            referer = request.headers.get("Referer")
+            if referer:
+                return RedirectResponse(referer)
+            else:
+                return RedirectResponse(request.url_for("admin:list", identity=self.identity))
+
+    admin.add_view(UserAdmin)
+    ```

sqladmin/__init__.py

@@ -1,11 +1,12 @@
-from sqladmin.application import Admin, expose
+from sqladmin.application import Admin, action, expose
 from sqladmin.models import BaseView, ModelAdmin, ModelView
 
 __version__ = "0.10.3"
 
 __all__ = [
     "Admin",
     "expose",
+    "action",
     "BaseView",
     "ModelAdmin",
     "ModelView",

sqladmin/application.py

@@ -1,6 +1,7 @@
 import inspect
 import io
 import logging
+from types import MethodType
 from typing import (
     Any,
     Callable,
@@ -10,6 +11,7 @@
     Tuple,
     Type,
     Union,
+    cast,
     no_type_check,
 )
 
@@ -30,11 +32,13 @@
 from sqladmin._types import ENGINE_TYPE
 from sqladmin.ajax import QueryAjaxModelLoader
 from sqladmin.authentication import AuthenticationBackend, login_required
+from sqladmin.helpers import slugify_action_name
 from sqladmin.models import BaseView, ModelView
 
 __all__ = [
     "Admin",
     "expose",
+    "action",
 ]
 
 logger = logging.getLogger(__name__)
@@ -117,6 +121,67 @@ def add_view(self, view: Union[Type[ModelView], Type[BaseView]]) -> None:
         else:
             self.add_base_view(view)
 
+    def _find_decorated_funcs(
+        self,
+        view: Type[Union[BaseView, ModelView]],
+        view_instance: Union[BaseView, ModelView],
+        handle_fn: Callable[
+            [MethodType, Type[Union[BaseView, ModelView]], Union[BaseView, ModelView]],
+            None,
+        ],
+    ) -> None:
+        funcs = inspect.getmembers(view_instance, predicate=inspect.ismethod)
+
+        for _, func in funcs[::-1]:
+            handle_fn(func, view, view_instance)
+
+    def _handle_action_decorated_func(
+        self,
+        func: MethodType,
+        view: Type[Union[BaseView, ModelView]],
+        view_instance: Union[BaseView, ModelView],
+    ) -> None:
+        if hasattr(func, "_action"):
+            view_instance = cast(ModelView, view_instance)
+            self.admin.add_route(
+                route=func,
+                path="/{identity}/action/" + getattr(func, "_slug"),
+                methods=["GET"],
+                name=f"{view_instance.identity}-{getattr(func, '_slug')}",
+                include_in_schema=getattr(func, "_include_in_schema"),
+            )
+
+            if getattr(func, "_add_in_list"):
+                view_instance._custom_actions_in_list[getattr(func, "_slug")] = getattr(
+                    func, "_label"
+                )
+            if getattr(func, "_add_in_detail"):
+                view_instance._custom_actions_in_detail[
+                    getattr(func, "_slug")
+                ] = getattr(func, "_label")
+
+            if getattr(func, "_confirmation_message"):
+                view_instance._custom_actions_confirmation[
+                    getattr(func, "_slug")
+                ] = getattr(func, "_confirmation_message")
+
+    def _handle_expose_decorated_func(
+        self,
+        func: MethodType,
+        view: Type[Union[BaseView, ModelView]],
+        view_instance: Union[BaseView, ModelView],
+    ) -> None:
+        if hasattr(func, "_exposed"):
+            self.admin.add_route(
+                route=func,
+                path=getattr(func, "_path"),
+                methods=getattr(func, "_methods"),
+                name=getattr(func, "_identity"),
+                include_in_schema=getattr(func, "_include_in_schema"),
+            )
+
+            view.identity = getattr(func, "_identity")
+
     def add_model_view(self, view: Type[ModelView]) -> None:
         """Add ModelView to the Admin.
 
@@ -152,7 +217,14 @@ class UserAdmin(ModelView, model=User):
             )
             view.async_engine = True
 
-        self._views.append((view()))
+        view_instance = view()
+
+        self._find_decorated_funcs(
+            view, view_instance, self._handle_action_decorated_func
+        )
+
+        view.templates = self.templates
+        self._views.append((view_instance))
 
     def add_base_view(self, view: Type[BaseView]) -> None:
         """Add BaseView to the Admin.
@@ -177,19 +249,10 @@ def test_page(self, request: Request):
         """
 
         view_instance = view()
-        funcs = inspect.getmembers(view_instance, predicate=inspect.ismethod)
-
-        for _, func in funcs[::-1]:
-            if hasattr(func, "_exposed"):
-                self.admin.add_route(
-                    route=func,
-                    path=func._path,
-                    methods=func._methods,
-                    name=func._identity,
-                    include_in_schema=func._include_in_schema,
-                )
 
-                view.identity = func._identity
+        self._find_decorated_funcs(
+            view, view_instance, self._handle_expose_decorated_func
+        )
 
         view.templates = self.templates
         self._views.append(view_instance)
@@ -638,3 +701,45 @@ def wrap(func):
         return func
 
     return wrap
+
+
+def action(
+    name: str,
+    label: Optional[str] = None,
+    confirmation_message: Optional[str] = None,
+    *,
+    include_in_schema: bool = True,
+    add_in_detail: bool = True,
+    add_in_list: bool = True,
+) -> Callable[..., Any]:
+    """Decorate a [`ModelView`][sqladmin.models.ModelView] function
+    with this to:
+
+    * expose it as a custom "action" route
+    * add a button to the admin panel to invoke the action
+
+    When invoked from the admin panel, the following query parameter(s) are passed:
+
+    * `pks`: the comma-separated list of selected object PKs - can be empty
+
+    Args:
+        name: Unique name for the action - must match `^[A-Za-z0-9 \-_]+$` regex
+        label: Human-readable text describing action
+        confirmation_message: Message to show before confirming action
+        include_in_schema: Should the endpoint be included in the schema?
+        add_in_detail: Should action be invocable from the "Detail" view?
+        add_in_list: Should action be invocable from the "List" view?
+    """
+
+    @no_type_check
+    def wrap(func):
+        func._action = True
+        func._slug = slugify_action_name(name)
+        func._label = label if label is not None else name
+        func._confirmation_message = confirmation_message
+        func._include_in_schema = include_in_schema
+        func._add_in_detail = add_in_detail
+        func._add_in_list = add_in_list
+        return func
+
+    return wrap

sqladmin/helpers.py

@@ -93,6 +93,16 @@ def slugify_class_name(name: str) -> str:
     return re.sub("([a-z0-9])([A-Z])", r"\1-\2", dashed).lower()
 
 
+def slugify_action_name(name: str) -> str:
+    if not re.search(r"^[A-Za-z0-9 \-_]+$", name):
+        raise ValueError(
+            "name must be non-empty and contain only allowed characters"
+            " - use `label` for more expressive names"
+        )
+
+    return re.sub(r"[_ ]", "-", name).lower()
+
+
 def secure_filename(filename: str) -> str:
     """Ported from Werkzeug.
 

sqladmin/models.py

@@ -724,6 +724,10 @@ def __init__(self) -> None:
                 model_admin=self, name=name, options=options
             )
 
+        self._custom_actions_in_list: Dict[str, str] = {}
+        self._custom_actions_in_detail: Dict[str, str] = {}
+        self._custom_actions_confirmation: Dict[str, str] = {}
+
     def _run_query_sync(self, stmt: ClauseElement) -> Any:
         with self.sessionmaker(expire_on_commit=False) as session:
             result = session.execute(stmt)
@@ -738,23 +742,23 @@ async def _run_query(self, stmt: ClauseElement) -> Any:
             return await anyio.to_thread.run_sync(self._run_query_sync, stmt)
 
     def _url_for_details(self, request: Request, obj: Any) -> Union[str, URL]:
-        pk = getattr(obj, get_primary_key(obj).name)
+        pk = self._get_pk(obj)
         return request.url_for(
             "admin:details",
             identity=slugify_class_name(obj.__class__.__name__),
             pk=pk,
         )
 
     def _url_for_edit(self, request: Request, obj: Any) -> Union[str, URL]:
-        pk = getattr(obj, get_primary_key(obj).name)
+        pk = self._get_pk(obj)
         return request.url_for(
             "admin:edit",
             identity=slugify_class_name(obj.__class__.__name__),
             pk=pk,
         )
 
     def _url_for_delete(self, request: Request, obj: Any) -> str:
-        pk = getattr(obj, get_primary_key(obj).name)
+        pk = self._get_pk(obj)
         query_params = urlencode({"pks": pk})
         url = request.url_for(
             "admin:delete", identity=slugify_class_name(obj.__class__.__name__)
@@ -775,6 +779,17 @@ def _url_for_details_with_prop(
             pk=pk,
         )
 
+    def _url_for_action(self, request: Request, action_name: str) -> str:
+        return str(
+            request.url_for(
+                f"admin:{self.identity}-{action_name}",
+                identity=self.identity,
+            )
+        )
+
+    def _get_pk(self, obj: Any) -> Any:
+        return getattr(obj, get_primary_key(obj).name)
+
     def _get_default_sort(self) -> List[Tuple[str, bool]]:
         if self.column_default_sort:
             if isinstance(self.column_default_sort, list):

sqladmin/statics/js/main.js

@@ -134,6 +134,17 @@ $("#action-delete").click(function () {
   });
 });
 
+$("[id^='action-custom-']").click(function () {
+  var pks = [];
+  $('.select-box').each(function () {
+    if ($(this).is(':checked')) {
+      pks.push($(this).siblings().get(0).value);
+    }
+  });
+
+  window.location.href = $(this).attr('data-url') + '?pks=' + pks.join(",");
+});
+
 // Select2 Tags
 $(':input[data-role="select2-tags"]').each(function () {
   $(this).select2({

sqladmin/templates/details.html

@@ -58,6 +58,19 @@ <h3 class="card-title">{{ model_view.pk_column.name }}: {{ model_view.get_prop_v
             </a>
           </div>
           {% endif %}
+          {% for custom_action,label in model_view._custom_actions_in_detail.items() %}
+          <div class="col-md-1">
+            {% if custom_action in model_view._custom_actions_confirmation %}
+            <a href="#" class="btn btn-secondary" data-bs-toggle="modal" data-bs-target="#modal-confirmation-{{ custom_action }}">
+              {{ label }}
+            </a>
+            {% else %}
+            <a href="{{ model_view._url_for_action(request, custom_action) }}?pks={{ model_view._get_pk(model) | string }}" class="btn btn-secondary">
+              {{ label }}
+            </a>
+            {% endif %}
+          </div>
+          {% endfor %}
         </div>
       </div>
     </div>
@@ -66,4 +79,13 @@ <h3 class="card-title">{{ model_view.pk_column.name }}: {{ model_view.get_prop_v
 {% if model_view.can_delete %}
 {% include 'modals/delete.html' %}
 {% endif %}
+
+{% for custom_action in model_view._custom_actions_in_detail %}
+{% if custom_action in model_view._custom_actions_confirmation %}
+{% with confirmation_message = model_view._custom_actions_confirmation[custom_action], custom_action=custom_action, url=model_view._url_for_action(request, custom_action) + '?pks=' + (model_view._get_pk(model) | string) %}
+{% include 'modals/details_action_confirmation.html' %}
+{% endwith %}
+{% endif %}
+{% endfor %}
+
 {% endblock %}