|  | 
|  | 1 | +from __future__ import annotations | 
|  | 2 | + | 
|  | 3 | +from dataclasses import dataclass | 
|  | 4 | +from typing import Any | 
|  | 5 | + | 
|  | 6 | +from django.conf import settings | 
|  | 7 | +from django.contrib import messages | 
|  | 8 | +from django.contrib.admin.options import BaseModelAdmin | 
|  | 9 | +from django.contrib.admin.templatetags.admin_urls import add_preserved_filters | 
|  | 10 | +from django.core.exceptions import FieldDoesNotExist | 
|  | 11 | +from django.http import HttpRequest | 
|  | 12 | +from django.http import HttpResponse | 
|  | 13 | +from django.http import HttpResponseRedirect | 
|  | 14 | +from django.utils.translation import gettext_lazy as _ | 
|  | 15 | + | 
|  | 16 | +import django_fsm as fsm | 
|  | 17 | + | 
|  | 18 | + | 
|  | 19 | +@dataclass | 
|  | 20 | +class FSMObjectTransition: | 
|  | 21 | +    fsm_field: str | 
|  | 22 | +    block_label: str | 
|  | 23 | +    available_transitions: list[fsm.Transition] | 
|  | 24 | + | 
|  | 25 | + | 
|  | 26 | +class FSMAdminMixin(BaseModelAdmin): | 
|  | 27 | +    change_form_template: str = "django_fsm/fsm_admin_change_form.html" | 
|  | 28 | + | 
|  | 29 | +    fsm_fields: list[str] = [] | 
|  | 30 | +    fsm_transition_success_msg = _("FSM transition '{transition_name}' succeeded.") | 
|  | 31 | +    fsm_transition_error_msg = _("FSM transition '{transition_name}' failed: {error}.") | 
|  | 32 | +    fsm_transition_not_allowed_msg = _("FSM transition '{transition_name}' is not allowed.") | 
|  | 33 | +    fsm_transition_not_valid_msg = _("FSM transition '{transition_name}' is not a valid.") | 
|  | 34 | +    fsm_context_key = "fsm_object_transitions" | 
|  | 35 | +    fsm_post_param = "_fsm_transition_to" | 
|  | 36 | +    default_disallow_transition = not getattr(settings, "FSM_ADMIN_FORCE_PERMIT", False) | 
|  | 37 | + | 
|  | 38 | +    def get_fsm_field_instance(self, fsm_field_name: str) -> fsm.FSMField | None: | 
|  | 39 | +        try: | 
|  | 40 | +            return self.model._meta.get_field(fsm_field_name) | 
|  | 41 | +        except FieldDoesNotExist: | 
|  | 42 | +            return None | 
|  | 43 | + | 
|  | 44 | +    def get_readonly_fields(self, request: HttpRequest, obj: Any = None) -> tuple[str]: | 
|  | 45 | +        read_only_fields = super().get_readonly_fields(request, obj) | 
|  | 46 | + | 
|  | 47 | +        for fsm_field_name in self.fsm_fields: | 
|  | 48 | +            if fsm_field_name in read_only_fields: | 
|  | 49 | +                continue | 
|  | 50 | +            field = self.get_fsm_field_instance(fsm_field_name=fsm_field_name) | 
|  | 51 | +            if field and getattr(field, "protected", False): | 
|  | 52 | +                read_only_fields += (fsm_field_name,) | 
|  | 53 | + | 
|  | 54 | +        return read_only_fields | 
|  | 55 | + | 
|  | 56 | +    @staticmethod | 
|  | 57 | +    def get_fsm_block_label(fsm_field_name: str) -> str: | 
|  | 58 | +        return f"Transition ({fsm_field_name})" | 
|  | 59 | + | 
|  | 60 | +    def get_fsm_object_transitions(self, request: HttpRequest, obj: Any) -> list[FSMObjectTransition]: | 
|  | 61 | +        fsm_object_transitions = [] | 
|  | 62 | + | 
|  | 63 | +        for field_name in sorted(self.fsm_fields): | 
|  | 64 | +            if func := getattr(obj, f"get_available_user_{field_name}_transitions"): | 
|  | 65 | +                fsm_object_transitions.append(  # noqa: PERF401 | 
|  | 66 | +                    FSMObjectTransition( | 
|  | 67 | +                        fsm_field=field_name, | 
|  | 68 | +                        block_label=self.get_fsm_block_label(fsm_field_name=field_name), | 
|  | 69 | +                        available_transitions=[ | 
|  | 70 | +                            t for t in func(user=request.user) if t.custom.get("admin", self.default_disallow_transition) | 
|  | 71 | +                        ], | 
|  | 72 | +                    ) | 
|  | 73 | +                ) | 
|  | 74 | + | 
|  | 75 | +        return fsm_object_transitions | 
|  | 76 | + | 
|  | 77 | +    def change_view( | 
|  | 78 | +        self, | 
|  | 79 | +        request: HttpRequest, | 
|  | 80 | +        object_id: str, | 
|  | 81 | +        form_url: str = "", | 
|  | 82 | +        extra_context: dict[str, Any] | None = None, | 
|  | 83 | +    ) -> HttpResponse: | 
|  | 84 | +        _context = extra_context or {} | 
|  | 85 | +        _context[self.fsm_context_key] = self.get_fsm_object_transitions( | 
|  | 86 | +            request=request, | 
|  | 87 | +            obj=self.get_object(request=request, object_id=object_id), | 
|  | 88 | +        ) | 
|  | 89 | + | 
|  | 90 | +        return super().change_view( | 
|  | 91 | +            request=request, | 
|  | 92 | +            object_id=object_id, | 
|  | 93 | +            form_url=form_url, | 
|  | 94 | +            extra_context=_context, | 
|  | 95 | +        ) | 
|  | 96 | + | 
|  | 97 | +    def get_fsm_redirect_url(self, request: HttpRequest, obj: Any) -> str: | 
|  | 98 | +        return request.path | 
|  | 99 | + | 
|  | 100 | +    def get_fsm_response(self, request: HttpRequest, obj: Any) -> HttpResponse: | 
|  | 101 | +        redirect_url = self.get_fsm_redirect_url(request=request, obj=obj) | 
|  | 102 | +        redirect_url = add_preserved_filters( | 
|  | 103 | +            context={ | 
|  | 104 | +                "preserved_filters": self.get_preserved_filters(request), | 
|  | 105 | +                "opts": self.model._meta, | 
|  | 106 | +            }, | 
|  | 107 | +            url=redirect_url, | 
|  | 108 | +        ) | 
|  | 109 | +        return HttpResponseRedirect(redirect_to=redirect_url) | 
|  | 110 | + | 
|  | 111 | +    def response_change(self, request: HttpRequest, obj: Any) -> HttpResponse: | 
|  | 112 | +        if self.fsm_post_param in request.POST: | 
|  | 113 | +            try: | 
|  | 114 | +                transition_name = request.POST[self.fsm_post_param] | 
|  | 115 | +                transition_func = getattr(obj, transition_name) | 
|  | 116 | +            except AttributeError: | 
|  | 117 | +                self.message_user( | 
|  | 118 | +                    request=request, | 
|  | 119 | +                    message=self.fsm_transition_not_valid_msg.format( | 
|  | 120 | +                        transition_name=transition_name, | 
|  | 121 | +                    ), | 
|  | 122 | +                    level=messages.ERROR, | 
|  | 123 | +                ) | 
|  | 124 | +                return self.get_fsm_response( | 
|  | 125 | +                    request=request, | 
|  | 126 | +                    obj=obj, | 
|  | 127 | +                ) | 
|  | 128 | + | 
|  | 129 | +            try: | 
|  | 130 | +                transition_func() | 
|  | 131 | +            except fsm.TransitionNotAllowed: | 
|  | 132 | +                self.message_user( | 
|  | 133 | +                    request=request, | 
|  | 134 | +                    message=self.fsm_transition_not_allowed_msg.format( | 
|  | 135 | +                        transition_name=transition_name, | 
|  | 136 | +                    ), | 
|  | 137 | +                    level=messages.ERROR, | 
|  | 138 | +                ) | 
|  | 139 | +            except fsm.ConcurrentTransition as err: | 
|  | 140 | +                self.message_user( | 
|  | 141 | +                    request=request, | 
|  | 142 | +                    message=self.fsm_transition_error_msg.format(transition_name=transition_name, error=str(err)), | 
|  | 143 | +                    level=messages.ERROR, | 
|  | 144 | +                ) | 
|  | 145 | +            else: | 
|  | 146 | +                obj.save() | 
|  | 147 | +                self.message_user( | 
|  | 148 | +                    request=request, | 
|  | 149 | +                    message=self.fsm_transition_success_msg.format( | 
|  | 150 | +                        transition_name=transition_name, | 
|  | 151 | +                    ), | 
|  | 152 | +                    level=messages.INFO, | 
|  | 153 | +                ) | 
|  | 154 | + | 
|  | 155 | +            return self.get_fsm_response( | 
|  | 156 | +                request=request, | 
|  | 157 | +                obj=obj, | 
|  | 158 | +            ) | 
|  | 159 | + | 
|  | 160 | +        return super().response_change(request=request, obj=obj) | 
0 commit comments