Step 7

Author: pfouque

.mypy.ini

@@ -50,8 +50,9 @@ repos:
       - id: ruff
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.7.0
+    rev: v1.7.1
     hooks:
       - id: mypy
         additional_dependencies:
           - django-stubs==4.2.6
+          - django-guardian

.pre-commit-config.yaml

@@ -5,13 +5,20 @@
 from __future__ import annotations
 
 import inspect
+from collections.abc import Callable
+from collections.abc import Collection
+from collections.abc import Generator
+from collections.abc import Iterable
+from collections.abc import Sequence
 from functools import partialmethod
 from functools import wraps
 from typing import TYPE_CHECKING
+from typing import Any
 
 from django.apps import apps as django_apps
 from django.db import models
 from django.db.models import Field
+from django.db.models import QuerySet
 from django.db.models.query_utils import DeferredAttribute
 from django.db.models.signals import class_prepared
 
@@ -34,30 +41,29 @@
 ]
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
-    from collections.abc import Generator
-    from collections.abc import Iterable
-    from collections.abc import Sequence
-    from typing import Any
+    from typing import Self
 
+    from _typeshed import Incomplete
     from django.contrib.auth.models import PermissionsMixin as UserWithPermissions
     from django.utils.functional import _StrOrPromise
 
-    _Model = models.Model
+    _FSMModel = models.Model
     _Field = models.Field[Any, Any]
     CharField = models.CharField[Any, Any]
     IntegerField = models.IntegerField[Any, Any]
     ForeignKey = models.ForeignKey[Any, Any]
 
     _StateValue = str | int
+    _Permission = str | Callable[[_FSMModel, UserWithPermissions], bool]
     _Instance = models.Model  # TODO: use real type
-    _ToDo = Any  # TODO: use real type
+
 else:
-    _Model = object
+    _FSMModel = object
     _Field = object
     CharField = models.CharField
     IntegerField = models.IntegerField
     ForeignKey = models.ForeignKey
+    Self = Any
 
 
 class TransitionNotAllowed(Exception):
@@ -277,7 +283,7 @@ class FSMFieldMixin(_Field):
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.protected = kwargs.pop("protected", False)
-        self.transitions: dict[type[_Model], dict[str, Any]] = {}  # cls -> (transitions name -> method)
+        self.transitions: dict[type[_FSMModel], dict[str, Any]] = {}  # cls -> (transitions name -> method)
         self.state_proxy = {}  # state -> ProxyClsRef
 
         state_choices = kwargs.pop("state_choices", None)
@@ -329,7 +335,7 @@ def set_proxy(self, instance: _Instance, state: str) -> None:
 
             instance.__class__ = model
 
-    def change_state(self, instance: _Instance, method: _ToDo, *args: Any, **kwargs: Any) -> Any:
+    def change_state(self, instance: _Instance, method: Incomplete, *args: Any, **kwargs: Any) -> Any:
         meta = method._django_fsm
         method_name = method.__name__
         current_state = self.get_state(instance)
@@ -382,7 +388,7 @@ def change_state(self, instance: _Instance, method: _ToDo, *args: Any, **kwargs:
 
         return result
 
-    def get_all_transitions(self, instance_cls: type[_Model]) -> Generator[Transition, None, None]:
+    def get_all_transitions(self, instance_cls: type[_FSMModel]) -> Generator[Transition, None, None]:
         """
         Returns [(source, target, name, method)] for all field transitions
         """
@@ -394,7 +400,7 @@ def get_all_transitions(self, instance_cls: type[_Model]) -> Generator[Transitio
             for transition in meta.transitions.values():
                 yield transition
 
-    def contribute_to_class(self, cls: type[_Model], name: str, private_only: bool = False, **kwargs: Any) -> None:
+    def contribute_to_class(self, cls: type[_FSMModel], name: str, private_only: bool = False, **kwargs: Any) -> None:
         self.base_cls = cls
 
         super().contribute_to_class(cls, name, private_only=private_only, **kwargs)
@@ -415,7 +421,7 @@ def _collect_transitions(self, *args: Any, **kwargs: Any) -> None:
         if not issubclass(sender, self.base_cls):
             return
 
-        def is_field_transition_method(attr: _ToDo) -> bool:
+        def is_field_transition_method(attr: Incomplete) -> bool:
             return (
                 (inspect.ismethod(attr) or inspect.isfunction(attr))
                 and hasattr(attr, "_django_fsm")
@@ -461,14 +467,14 @@ class FSMKeyField(FSMFieldMixin, ForeignKey):
     State Machine support for Django model
     """
 
-    def get_state(self, instance: _Instance) -> _ToDo:
+    def get_state(self, instance: _Instance) -> Incomplete:
         return instance.__dict__[self.attname]
 
     def set_state(self, instance: _Instance, state: str) -> None:
         instance.__dict__[self.attname] = self.to_python(state)
 
 
-class FSMModelMixin(_Model):
+class FSMModelMixin(_FSMModel):
     """
     Mixin that allows refresh_from_db for models with fsm protected fields
     """
@@ -495,7 +501,7 @@ def refresh_from_db(self, *args, **kwargs):
         super().refresh_from_db(*args, **kwargs)
 
 
-class ConcurrentTransitionMixin(_Model):
+class ConcurrentTransitionMixin(_FSMModel):
     """
     Protects a Model from undesirable effects caused by concurrently executed transitions,
     e.g. running the same transition multiple times at the same time, or running different
@@ -529,7 +535,15 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
     def state_fields(self) -> Iterable[Any]:
         return filter(lambda field: isinstance(field, FSMFieldMixin), self._meta.fields)
 
-    def _do_update(self, base_qs, using, pk_val, values, update_fields, forced_update):  # type: ignore[no-untyped-def]
+    def _do_update(
+        self,
+        base_qs: QuerySet[Self],
+        using: Any,
+        pk_val: Any,
+        values: Collection[Any] | None,
+        update_fields: Iterable[str] | None,
+        forced_update: bool,
+    ) -> bool:
         # _do_update is called once for each model class in the inheritance hierarchy.
         # We can only filter the base_qs on state fields (can be more than one!) present in this particular model.
 
@@ -539,7 +553,7 @@ def _do_update(self, base_qs, using, pk_val, values, update_fields, forced_updat
         # state filter will be used to narrow down the standard filter checking only PK
         state_filter = {field.attname: self.__initial_states[field.attname] for field in filter_on}
 
-        updated = super()._do_update(  # type: ignore[misc]
+        updated: bool = super()._do_update(  # type: ignore[misc]
             base_qs=base_qs.filter(**state_filter),
             using=using,
             pk_val=pk_val,
@@ -577,7 +591,7 @@ def transition(
     target: _StateValue | State | None = None,
     on_error: _StateValue | None = None,
     conditions: list[Callable[[Any], bool]] = [],
-    permission: str | Callable[[models.Model, UserWithPermissions], bool] | None = None,
+    permission: _Permission | None = None,
     custom: dict[str, _StrOrPromise] = {},
 ) -> Callable[[Any], Any]:
     """
@@ -587,21 +601,22 @@ def transition(
     has not changed after the function call.
     """
 
-    def inner_transition(func: _ToDo) -> _ToDo:
+    def inner_transition(func: Incomplete) -> Incomplete:
         wrapper_installed, fsm_meta = True, getattr(func, "_django_fsm", None)
         if not fsm_meta:
             wrapper_installed = False
             fsm_meta = FSMMeta(field=field, method=func)
             setattr(func, "_django_fsm", fsm_meta)
 
+        # if isinstance(source, Iterable):
         if isinstance(source, (list, tuple, set)):
             for state in source:
                 func._django_fsm.add_transition(func, state, target, on_error, conditions, permission, custom)
         else:
             func._django_fsm.add_transition(func, source, target, on_error, conditions, permission, custom)
 
         @wraps(func)
-        def _change_state(instance: _Instance, *args: Any, **kwargs: Any) -> _ToDo:
+        def _change_state(instance: _Instance, *args: Any, **kwargs: Any) -> Incomplete:
             return fsm_meta.field.change_state(instance, func, *args, **kwargs)
 
         if not wrapper_installed:
@@ -612,7 +627,7 @@ def _change_state(instance: _Instance, *args: Any, **kwargs: Any) -> _ToDo:
     return inner_transition
 
 
-def can_proceed(bound_method: _ToDo, check_conditions: bool = True) -> bool:
+def can_proceed(bound_method: Incomplete, check_conditions: bool = True) -> bool:
     """
     Returns True if model in state allows to call bound_method
 
@@ -629,7 +644,7 @@ def can_proceed(bound_method: _ToDo, check_conditions: bool = True) -> bool:
     return meta.has_transition(current_state) and (not check_conditions or meta.conditions_met(self, current_state))
 
 
-def has_transition_perm(bound_method: _ToDo, user: UserWithPermissions) -> bool:
+def has_transition_perm(bound_method: Incomplete, user: UserWithPermissions) -> bool:
     """
     Returns True if model in state allows to call bound_method and user have rights on it
     """
@@ -648,15 +663,15 @@ def has_transition_perm(bound_method: _ToDo, user: UserWithPermissions) -> bool:
 
 
 class State:
-    def get_state(self, model: _Model, transition: Transition, result: Any, args: Any = [], kwargs: Any = {}) -> _ToDo:
+    def get_state(self, model: _FSMModel, transition: Transition, result: Any, args: Any = [], kwargs: Any = {}) -> Incomplete:
         raise NotImplementedError
 
 
 class RETURN_VALUE(State):
     def __init__(self, *allowed_states: Sequence[_StateValue]) -> None:
         self.allowed_states = allowed_states if allowed_states else None
 
-    def get_state(self, model: _Model, transition: Transition, result: Any, args: Any = [], kwargs: Any = {}) -> _ToDo:
+    def get_state(self, model: _FSMModel, transition: Transition, result: Any, args: Any = [], kwargs: Any = {}) -> Incomplete:
         if self.allowed_states is not None:
             if result not in self.allowed_states:
                 raise InvalidResultState(f"{result} is not in list of allowed states\n{self.allowed_states}")
@@ -669,8 +684,8 @@ def __init__(self, func: Callable[..., _StateValue | Any], states: Sequence[_Sta
         self.allowed_states = states
 
     def get_state(
-        self, model: _Model, transition: Transition, result: _StateValue | Any, args: Any = [], kwargs: Any = {}
-    ) -> _ToDo:
+        self, model: _FSMModel, transition: Transition, result: _StateValue | Any, args: Any = [], kwargs: Any = {}
+    ) -> Incomplete:
         result_state = self.func(model, *args, **kwargs)
         if self.allowed_states is not None:
             if result_state not in self.allowed_states:

django_fsm/__init__.py

@@ -75,6 +75,79 @@ fixable = ["I"]
 force-single-line = true
 required-imports = ["from __future__ import annotations"]
 
+[tool.django-stubs]
+django_settings_module = "tests.settings"
+
+[tool.mypy]
+python_version = 3.11
+plugins = ["mypy_django_plugin.main"]
+
+# Start off with these
+warn_unused_configs = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+
+# Getting these passing should be easy
+strict_equality = true
+extra_checks = true
+
+# Strongly recommend enabling this one as soon as you can
+check_untyped_defs = true
+
+# These shouldn't be too much additional work, but may be tricky to
+# get passing if you use a lot of untyped libraries
+disallow_subclassing_any = true
+disallow_untyped_decorators = true
+disallow_any_generics = true
+
+# These next few are various gradations of forcing use of type annotations
+disallow_untyped_calls = true
+disallow_incomplete_defs = true
+disallow_untyped_defs = true
+
+# This one isn't too hard to get passing, but return on investment is lower
+no_implicit_reexport = true
+
+# This one can be tricky to get passing if you use a lot of untyped libraries
+warn_return_any = true
+
+[[tool.mypy.overrides]]
+module = [
+    "tests.*",
+    "django_fsm.tests.*"
+]
+ignore_errors = true
+
+# Start off with these
+warn_unused_ignores = true
+
+# Getting these passing should be easy
+strict_equality = false
+extra_checks = false
+
+# Strongly recommend enabling this one as soon as you can
+check_untyped_defs = false
+# These shouldn't be too much additional work, but may be tricky to
+# get passing if you use a lot of untyped libraries
+disallow_subclassing_any = false
+disallow_untyped_decorators = false
+disallow_any_generics = false
+
+# These next few are various gradations of forcing use of type annotations
+disallow_untyped_calls = false
+disallow_incomplete_defs = false
+disallow_untyped_defs = false
+
+# This one isn't too hard to get passing, but return on investment is lower
+no_implicit_reexport = false
+
+# This one can be tricky to get passing if you use a lot of untyped libraries
+warn_return_any = false
+
+[[tool.mypy.overrides]]
+module = "django_fsm.management.commands.graph_transitions"
+ignore_errors = true
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

pyproject.toml

@@ -93,7 +93,7 @@ class BlogPost(models.Model):
 
     state = FSMField(default="new", protected=True)
 
-    def can_restore(self, user):
+    def can_restore(self, user) -> bool:
         return user.is_superuser or user.is_staff
 
     @transition(field=state, source="new", target="published", on_error="failed", permission="testapp.can_publish_post")

tests/testapp/models.py

@@ -8,7 +8,7 @@
 from django_fsm.signals import post_transition
 
 
-class TestModel(models.Model):
+class MultipletransitionsModel(models.Model):
     counter = models.IntegerField(default=0)
     signal_counter = models.IntegerField(default=0)
     state = FSMField(default="SUBMITTED_BY_USER")
@@ -27,12 +27,12 @@ def count_calls(sender, instance, name, source, target, **kwargs):
     instance.signal_counter += 1
 
 
-post_transition.connect(count_calls, sender=TestModel)
+post_transition.connect(count_calls, sender=MultipletransitionsModel)
 
 
 class TestStateProxy(TestCase):
     def test_transition_method_called_once(self):
-        model = TestModel()
+        model = MultipletransitionsModel()
         model.review()
         self.assertEqual(1, model.counter)
         self.assertEqual(1, model.signal_counter)