From b30c4be1d6e879a912530e0b5c1e9882294903f7 Mon Sep 17 00:00:00 2001 From: Pascal F Date: Tue, 21 Nov 2023 21:54:57 +0100 Subject: [PATCH 01/10] Setup mypy --- .mypy.ini | 38 ++++++++++++++++++++++++++++++++++++++ .pre-commit-config.yaml | 7 +++++++ django_fsm/__init__.py | 4 ++-- 3 files changed, 47 insertions(+), 2 deletions(-) create mode 100644 .mypy.ini diff --git a/.mypy.ini b/.mypy.ini new file mode 100644 index 0000000..8ee1bde --- /dev/null +++ b/.mypy.ini @@ -0,0 +1,38 @@ +[mypy] +# 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 + +[mypy-tests.*] +ignore_errors = True + +[mypy-django_fsm.tests.*] +ignore_errors = True + +[mypy-django_fsm.management.commands.graph_transitions] +ignore_errors = True diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index d1822d3..1bf2d2a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -48,3 +48,10 @@ repos: hooks: - id: ruff-format - id: ruff + + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.7.0 + hooks: + - id: mypy + additional_dependencies: + - django-stubs==4.2.6 diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index aa0779e..b047562 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -347,10 +347,10 @@ def get_all_transitions(self, instance_cls): for transition in meta.transitions.values(): yield transition - def contribute_to_class(self, cls, name, **kwargs): + def contribute_to_class(self, cls, name, private_only=False, **kwargs): self.base_cls = cls - super().contribute_to_class(cls, name, **kwargs) + super().contribute_to_class(cls, name, private_only=private_only, **kwargs) setattr(cls, self.name, self.descriptor_class(self)) setattr(cls, f"get_all_{self.name}_transitions", partialmethod(get_all_FIELD_transitions, field=self)) setattr(cls, f"get_available_{self.name}_transitions", partialmethod(get_available_FIELD_transitions, field=self)) From 5c64979429bd7b5e6a1bbbe3addffed79affbd82 Mon Sep 17 00:00:00 2001 From: Pascal F Date: Tue, 21 Nov 2023 22:02:12 +0100 Subject: [PATCH 02/10] Enable already passing rules --- .mypy.ini | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/.mypy.ini b/.mypy.ini index 8ee1bde..629535a 100644 --- a/.mypy.ini +++ b/.mypy.ini @@ -1,20 +1,20 @@ [mypy] # Start off with these -; warn_unused_configs = True -; warn_redundant_casts = True -; warn_unused_ignores = True +warn_unused_configs = True +warn_redundant_casts = True +warn_unused_ignores = True # Getting these passing should be easy -; strict_equality = True -; extra_checks = True +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_subclassing_any = True +disallow_untyped_decorators = True ; disallow_any_generics = True # These next few are various gradations of forcing use of type annotations @@ -23,10 +23,10 @@ ; disallow_untyped_defs = True # This one isn't too hard to get passing, but return on investment is lower -; no_implicit_reexport = True +no_implicit_reexport = True # This one can be tricky to get passing if you use a lot of untyped libraries -; warn_return_any = True +warn_return_any = True [mypy-tests.*] ignore_errors = True From 34ccb22c8dd84ce682b85931ee3a8fdfeef66a0f Mon Sep 17 00:00:00 2001 From: Pascal F Date: Wed, 22 Nov 2023 05:10:53 +0100 Subject: [PATCH 03/10] step 1 --- .mypy.ini | 2 +- django_fsm/__init__.py | 64 ++++++++++++++++++++++++------------------ 2 files changed, 37 insertions(+), 29 deletions(-) diff --git a/.mypy.ini b/.mypy.ini index 629535a..0117be4 100644 --- a/.mypy.ini +++ b/.mypy.ini @@ -9,7 +9,7 @@ strict_equality = True extra_checks = True # Strongly recommend enabling this one as soon as you can -; check_untyped_defs = True +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 diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index b047562..f08f3e2 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -7,6 +7,9 @@ import inspect from functools import partialmethod from functools import wraps +from typing import TYPE_CHECKING +from typing import Any +from typing import Callable from django.apps import apps as django_apps from django.db import models @@ -32,11 +35,16 @@ "RETURN_VALUE", ] +if TYPE_CHECKING: + _Model = models.Model +else: + _Model = object + class TransitionNotAllowed(Exception): """Raised when a transition is not allowed""" - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs) -> None: self.object = kwargs.pop("object", None) self.method = kwargs.pop("method", None) super().__init__(*args, **kwargs) @@ -55,7 +63,7 @@ class ConcurrentTransition(Exception): class Transition: - def __init__(self, method, source, target, on_error, conditions, permission, custom): + def __init__(self, method: Callable, source, target, on_error, conditions, permission, custom) -> None: self.method = method self.source = source self.target = target @@ -65,10 +73,10 @@ def __init__(self, method, source, target, on_error, conditions, permission, cus self.custom = custom @property - def name(self): + def name(self) -> str: return self.method.__name__ - def has_perm(self, instance, user): + def has_perm(self, instance, user) -> bool: if not self.permission: return True if callable(self.permission): @@ -127,9 +135,9 @@ class FSMMeta: Models methods transitions meta information """ - def __init__(self, field, method): + def __init__(self, field, method) -> None: self.field = field - self.transitions = {} # source -> Transition + self.transitions: dict[str, Any] = {} # source -> Transition def get_transition(self, source): transition = self.transitions.get(source, None) @@ -139,7 +147,7 @@ def get_transition(self, source): transition = self.transitions.get("+", None) return transition - def add_transition(self, method, source, target, on_error=None, conditions=[], permission=None, custom={}): + def add_transition(self, method, source, target, on_error=None, conditions=[], permission=None, custom={}) -> None: if source in self.transitions: raise AssertionError(f"Duplicate transition for {source} state") @@ -153,7 +161,7 @@ def add_transition(self, method, source, target, on_error=None, conditions=[], p custom=custom, ) - def has_transition(self, state): + def has_transition(self, state) -> bool: """ Lookup if any transition exists from current model state using current method """ @@ -168,7 +176,7 @@ def has_transition(self, state): return False - def conditions_met(self, instance, state): + def conditions_met(self, instance, state) -> bool: """ Check if all conditions have been met """ @@ -182,13 +190,13 @@ def conditions_met(self, instance, state): return all(condition(instance) for condition in transition.conditions) - def has_transition_perm(self, instance, state, user): + def has_transition_perm(self, instance, state, user) -> bool: transition = self.get_transition(state) if not transition: return False - - return transition.has_perm(instance, user) + else: + return bool(transition.has_perm(instance, user)) def next_state(self, current_state): transition = self.get_transition(current_state) @@ -208,7 +216,7 @@ def exception_state(self, current_state): class FSMFieldDescriptor: - def __init__(self, field): + def __init__(self, field) -> None: self.field = field def __get__(self, instance, type=None): @@ -216,7 +224,7 @@ def __get__(self, instance, type=None): return self return self.field.get_state(instance) - def __set__(self, instance, value): + def __set__(self, instance, value) -> None: if self.field.protected and self.field.name in instance.__dict__: raise AttributeError(f"Direct {self.field.name} modification is not allowed") @@ -225,12 +233,12 @@ def __set__(self, instance, value): self.field.set_state(instance, value) -class FSMFieldMixin: +class FSMFieldMixin(Field): descriptor_class = FSMFieldDescriptor - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs) -> None: self.protected = kwargs.pop("protected", False) - self.transitions = {} # cls -> (transitions name -> method) + self.transitions: dict[Any, dict[str, Any]] = {} # cls -> (transitions name -> method) self.state_proxy = {} # state -> ProxyClsRef state_choices = kwargs.pop("state_choices", None) @@ -256,7 +264,7 @@ def deconstruct(self): def get_state(self, instance): # The state field may be deferred. We delegate the logic of figuring this out # and loading the deferred field on-demand to Django's built-in DeferredAttribute class. - return DeferredAttribute(self).__get__(instance) + return DeferredAttribute(self).__get__(instance) # type: ignore[attr-defined] def set_state(self, instance, state): instance.__dict__[self.name] = state @@ -396,7 +404,7 @@ class FSMField(FSMFieldMixin, models.CharField): State Machine support for Django model as CharField """ - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs) -> None: kwargs.setdefault("max_length", 50) super().__init__(*args, **kwargs) @@ -421,7 +429,7 @@ def set_state(self, instance, state): instance.__dict__[self.attname] = self.to_python(state) -class FSMModelMixin: +class FSMModelMixin(_Model): """ Mixin that allows refresh_from_db for models with fsm protected fields """ @@ -448,7 +456,7 @@ def refresh_from_db(self, *args, **kwargs): super().refresh_from_db(*args, **kwargs) -class ConcurrentTransitionMixin: +class ConcurrentTransitionMixin(_Model): """ 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 @@ -474,7 +482,7 @@ class ConcurrentTransitionMixin: state, thus practically negating their effect. """ - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) self._update_initial_state() @@ -492,7 +500,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( + updated = super()._do_update( # type: ignore[misc] base_qs=base_qs.filter(**state_filter), using=using, pk_val=pk_val, @@ -557,7 +565,7 @@ def _change_state(instance, *args, **kwargs): return inner_transition -def can_proceed(bound_method, check_conditions=True): +def can_proceed(bound_method, check_conditions=True) -> bool: """ Returns True if model in state allows to call bound_method @@ -574,7 +582,7 @@ def can_proceed(bound_method, check_conditions=True): return meta.has_transition(current_state) and (not check_conditions or meta.conditions_met(self, current_state)) -def has_transition_perm(bound_method, user): +def has_transition_perm(bound_method, user) -> bool: """ Returns True if model in state allows to call bound_method and user have rights on it """ @@ -585,7 +593,7 @@ def has_transition_perm(bound_method, user): self = bound_method.__self__ current_state = meta.field.get_state(self) - return ( + return bool( meta.has_transition(current_state) and meta.conditions_met(self, current_state) and meta.has_transition_perm(self, current_state, user) @@ -598,7 +606,7 @@ def get_state(self, model, transition, result, args=[], kwargs={}): class RETURN_VALUE(State): - def __init__(self, *allowed_states): + def __init__(self, *allowed_states) -> None: self.allowed_states = allowed_states if allowed_states else None def get_state(self, model, transition, result, args=[], kwargs={}): @@ -609,7 +617,7 @@ def get_state(self, model, transition, result, args=[], kwargs={}): class GET_STATE(State): - def __init__(self, func, states=None): + def __init__(self, func, states=None) -> None: self.func = func self.allowed_states = states From ade2c832a84cd055e589d1fab889a706e23a39f4 Mon Sep 17 00:00:00 2001 From: Pascal F Date: Wed, 22 Nov 2023 06:30:39 +0100 Subject: [PATCH 04/10] Step 2 --- django_fsm/__init__.py | 30 ++++++++++++++++++++++++++---- 1 file changed, 26 insertions(+), 4 deletions(-) diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index f08f3e2..f7080ec 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -8,8 +8,6 @@ from functools import partialmethod from functools import wraps from typing import TYPE_CHECKING -from typing import Any -from typing import Callable from django.apps import apps as django_apps from django.db import models @@ -36,6 +34,13 @@ ] if TYPE_CHECKING: + from collections.abc import Callable + from collections.abc import Sequence + from typing import Any + + from django.contrib.auth.models import AbstractBaseUser + from django.utils.functional import _StrOrPromise + _Model = models.Model else: _Model = object @@ -63,7 +68,16 @@ class ConcurrentTransition(Exception): class Transition: - def __init__(self, method: Callable, source, target, on_error, conditions, permission, custom) -> None: + def __init__( + self, + method: Callable, + source: str | int | Sequence[str | int] | State, + target: str | int | State | None, + on_error: str | int | None, + conditions: list[Callable[[Any], bool]], + permission: str | Callable[[models.Model, AbstractBaseUser], bool] | None, + custom: dict[str, _StrOrPromise], + ) -> None: self.method = method self.source = source self.target = target @@ -532,7 +546,15 @@ def save(self, *args, **kwargs): self._update_initial_state() -def transition(field, source="*", target=None, on_error=None, conditions=[], permission=None, custom={}): +def transition( + field, + source: str | int | Sequence[str | int] | State = "*", + target: str | int | State | None = None, + on_error: str | int | None = None, + conditions: list[Callable[[Any], bool]] = [], + permission: str | Callable[[models.Model, AbstractBaseUser], bool] | None = None, + custom: dict[str, _StrOrPromise] = {}, +): """ Method decorator to mark allowed transitions. From 9bf8fde9248c3bbd849a4e6d28d90ce3e4ee80aa Mon Sep 17 00:00:00 2001 From: Pascal F Date: Wed, 22 Nov 2023 07:30:45 +0100 Subject: [PATCH 05/10] Step 3 --- .mypy.ini | 4 +- django_fsm/__init__.py | 86 ++++++++++++++++++++++++++---------------- 2 files changed, 55 insertions(+), 35 deletions(-) diff --git a/.mypy.ini b/.mypy.ini index 0117be4..6cf1061 100644 --- a/.mypy.ini +++ b/.mypy.ini @@ -15,10 +15,10 @@ check_untyped_defs = True # get passing if you use a lot of untyped libraries disallow_subclassing_any = True disallow_untyped_decorators = True -; disallow_any_generics = True +disallow_any_generics = True # These next few are various gradations of forcing use of type annotations -; disallow_untyped_calls = True +disallow_untyped_calls = True ; disallow_incomplete_defs = True ; disallow_untyped_defs = True diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index f7080ec..9a7e809 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -35,21 +35,30 @@ if TYPE_CHECKING: from collections.abc import Callable + from collections.abc import Generator from collections.abc import Sequence from typing import Any - from django.contrib.auth.models import AbstractBaseUser + from django.contrib.auth.models import PermissionsMixin as UserWithPermissions from django.utils.functional import _StrOrPromise _Model = models.Model + _Field = models.Field[Any, Any] + CharField = models.CharField[str, str] + IntegerField = models.IntegerField[int, int] + ForeignKey = models.ForeignKey[Any, Any] else: _Model = object + _Field = object + CharField = models.CharField + IntegerField = models.IntegerField + ForeignKey = models.ForeignKey class TransitionNotAllowed(Exception): """Raised when a transition is not allowed""" - def __init__(self, *args, **kwargs) -> None: + def __init__(self, *args: Any, **kwargs: Any) -> None: self.object = kwargs.pop("object", None) self.method = kwargs.pop("method", None) super().__init__(*args, **kwargs) @@ -70,12 +79,12 @@ class ConcurrentTransition(Exception): class Transition: def __init__( self, - method: Callable, + method: Callable[..., Any], source: str | int | Sequence[str | int] | State, target: str | int | State | None, on_error: str | int | None, conditions: list[Callable[[Any], bool]], - permission: str | Callable[[models.Model, AbstractBaseUser], bool] | None, + permission: str | Callable[[models.Model, UserWithPermissions], bool] | None, custom: dict[str, _StrOrPromise], ) -> None: self.method = method @@ -90,7 +99,7 @@ def __init__( def name(self) -> str: return self.method.__name__ - def has_perm(self, instance, user) -> bool: + def has_perm(self, instance, user: UserWithPermissions) -> bool: if not self.permission: return True if callable(self.permission): @@ -113,7 +122,7 @@ def __eq__(self, other): return False -def get_available_FIELD_transitions(instance, field): +def get_available_FIELD_transitions(instance, field: FSMFieldMixin) -> Generator[Transition, None, None]: """ List of transitions available in current model state with all conditions met @@ -127,14 +136,16 @@ def get_available_FIELD_transitions(instance, field): yield meta.get_transition(curr_state) -def get_all_FIELD_transitions(instance, field): +def get_all_FIELD_transitions(instance, field: FSMFieldMixin) -> Generator[Transition, None, None]: """ List of all transitions available in current model state """ return field.get_all_transitions(instance.__class__) -def get_available_user_FIELD_transitions(instance, user, field): +def get_available_user_FIELD_transitions( + instance, user: UserWithPermissions, field: FSMFieldMixin +) -> Generator[Transition, None, None]: """ List of transitions available in current model state with all conditions met and user have rights on it @@ -153,7 +164,7 @@ def __init__(self, field, method) -> None: self.field = field self.transitions: dict[str, Any] = {} # source -> Transition - def get_transition(self, source): + def get_transition(self, source: str): transition = self.transitions.get(source, None) if transition is None: transition = self.transitions.get("*", None) @@ -161,7 +172,16 @@ def get_transition(self, source): transition = self.transitions.get("+", None) return transition - def add_transition(self, method, source, target, on_error=None, conditions=[], permission=None, custom={}) -> None: + def add_transition( + self, + method: Callable[..., Any], + source: str, + target: str | int, + on_error: str | int | None = None, + conditions: list[Callable[[Any], bool]] = [], + permission: str | Callable[[models.Model, UserWithPermissions], bool] | None = None, + custom: dict[str, _StrOrPromise] = {}, + ) -> None: if source in self.transitions: raise AssertionError(f"Duplicate transition for {source} state") @@ -204,7 +224,7 @@ def conditions_met(self, instance, state) -> bool: return all(condition(instance) for condition in transition.conditions) - def has_transition_perm(self, instance, state, user) -> bool: + def has_transition_perm(self, instance, state, user: UserWithPermissions) -> bool: transition = self.get_transition(state) if not transition: @@ -247,10 +267,10 @@ def __set__(self, instance, value) -> None: self.field.set_state(instance, value) -class FSMFieldMixin(Field): +class FSMFieldMixin(_Field): descriptor_class = FSMFieldDescriptor - def __init__(self, *args, **kwargs) -> None: + def __init__(self, *args: Any, **kwargs: Any) -> None: self.protected = kwargs.pop("protected", False) self.transitions: dict[Any, dict[str, Any]] = {} # cls -> (transitions name -> method) self.state_proxy = {} # state -> ProxyClsRef @@ -275,15 +295,15 @@ def deconstruct(self): kwargs["protected"] = self.protected return name, path, args, kwargs - def get_state(self, instance): + def get_state(self, instance) -> Any: # The state field may be deferred. We delegate the logic of figuring this out # and loading the deferred field on-demand to Django's built-in DeferredAttribute class. return DeferredAttribute(self).__get__(instance) # type: ignore[attr-defined] - def set_state(self, instance, state): + def set_state(self, instance, state: str) -> None: instance.__dict__[self.name] = state - def set_proxy(self, instance, state): + def set_proxy(self, instance, state: str) -> None: """ Change class """ @@ -304,7 +324,7 @@ def set_proxy(self, instance, state): instance.__class__ = model - def change_state(self, instance, method, *args, **kwargs): + def change_state(self, instance, method, *args: Any, **kwargs: Any): meta = method._django_fsm method_name = method.__name__ current_state = self.get_state(instance) @@ -357,7 +377,7 @@ def change_state(self, instance, method, *args, **kwargs): return result - def get_all_transitions(self, instance_cls): + def get_all_transitions(self, instance_cls) -> Generator[Transition, None, None]: """ Returns [(source, target, name, method)] for all field transitions """ @@ -384,7 +404,7 @@ def contribute_to_class(self, cls, name, private_only=False, **kwargs): class_prepared.connect(self._collect_transitions) - def _collect_transitions(self, *args, **kwargs): + def _collect_transitions(self, *args: Any, **kwargs: Any): sender = kwargs["sender"] if not issubclass(sender, self.base_cls): @@ -413,17 +433,17 @@ def is_field_transition_method(attr): self.transitions[sender] = sender_transitions -class FSMField(FSMFieldMixin, models.CharField): +class FSMField(FSMFieldMixin, CharField): """ State Machine support for Django model as CharField """ - def __init__(self, *args, **kwargs) -> None: + def __init__(self, *args: Any, **kwargs: Any) -> None: kwargs.setdefault("max_length", 50) super().__init__(*args, **kwargs) -class FSMIntegerField(FSMFieldMixin, models.IntegerField): +class FSMIntegerField(FSMFieldMixin, IntegerField): """ Same as FSMField, but stores the state value in an IntegerField. """ @@ -431,7 +451,7 @@ class FSMIntegerField(FSMFieldMixin, models.IntegerField): pass -class FSMKeyField(FSMFieldMixin, models.ForeignKey): +class FSMKeyField(FSMFieldMixin, ForeignKey): """ State Machine support for Django model """ @@ -496,7 +516,7 @@ class ConcurrentTransitionMixin(_Model): state, thus practically negating their effect. """ - def __init__(self, *args, **kwargs) -> None: + def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self._update_initial_state() @@ -534,14 +554,14 @@ def _do_update(self, base_qs, using, pk_val, values, update_fields, forced_updat return updated - def _update_initial_state(self): + def _update_initial_state(self) -> None: self.__initial_states = {field.attname: field.value_from_object(self) for field in self.state_fields} - def refresh_from_db(self, *args, **kwargs): + def refresh_from_db(self, *args: Any, **kwargs: Any) -> None: super().refresh_from_db(*args, **kwargs) self._update_initial_state() - def save(self, *args, **kwargs): + def save(self, *args: Any, **kwargs: Any) -> None: super().save(*args, **kwargs) self._update_initial_state() @@ -552,7 +572,7 @@ def transition( target: str | int | State | None = None, on_error: str | int | None = None, conditions: list[Callable[[Any], bool]] = [], - permission: str | Callable[[models.Model, AbstractBaseUser], bool] | None = None, + permission: str | Callable[[models.Model, UserWithPermissions], bool] | None = None, custom: dict[str, _StrOrPromise] = {}, ): """ @@ -576,7 +596,7 @@ def inner_transition(func): func._django_fsm.add_transition(func, source, target, on_error, conditions, permission, custom) @wraps(func) - def _change_state(instance, *args, **kwargs): + def _change_state(instance, *args: Any, **kwargs: Any): return fsm_meta.field.change_state(instance, func, *args, **kwargs) if not wrapper_installed: @@ -587,7 +607,7 @@ def _change_state(instance, *args, **kwargs): return inner_transition -def can_proceed(bound_method, check_conditions=True) -> bool: +def can_proceed(bound_method, check_conditions: bool = True) -> bool: """ Returns True if model in state allows to call bound_method @@ -604,7 +624,7 @@ def can_proceed(bound_method, check_conditions=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, user) -> bool: +def has_transition_perm(bound_method, user: UserWithPermissions) -> bool: """ Returns True if model in state allows to call bound_method and user have rights on it """ @@ -628,7 +648,7 @@ def get_state(self, model, transition, result, args=[], kwargs={}): class RETURN_VALUE(State): - def __init__(self, *allowed_states) -> None: + def __init__(self, *allowed_states: Sequence[str | int]) -> None: self.allowed_states = allowed_states if allowed_states else None def get_state(self, model, transition, result, args=[], kwargs={}): @@ -639,7 +659,7 @@ def get_state(self, model, transition, result, args=[], kwargs={}): class GET_STATE(State): - def __init__(self, func, states=None) -> None: + def __init__(self, func: Callable[..., str | int], states: Sequence[str | int] | None = None) -> None: self.func = func self.allowed_states = states From af123460463448a01d0cd49a685cf4ed8f14357d Mon Sep 17 00:00:00 2001 From: Pascal F Date: Wed, 22 Nov 2023 08:41:09 +0100 Subject: [PATCH 06/10] Step 4 --- .mypy.ini | 2 +- django_fsm/__init__.py | 82 ++++++++++++++++++++++-------------------- 2 files changed, 44 insertions(+), 40 deletions(-) diff --git a/.mypy.ini b/.mypy.ini index 6cf1061..75ef771 100644 --- a/.mypy.ini +++ b/.mypy.ini @@ -19,7 +19,7 @@ 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_incomplete_defs = True ; disallow_untyped_defs = True # This one isn't too hard to get passing, but return on investment is lower diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index 9a7e809..445c8bd 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -36,6 +36,7 @@ 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 @@ -47,6 +48,9 @@ CharField = models.CharField[str, str] IntegerField = models.IntegerField[int, int] ForeignKey = models.ForeignKey[Any, Any] + + _Instance = models.Model # TODO: use real type + _ToDo = Any # TODO: use real type else: _Model = object _Field = object @@ -79,12 +83,12 @@ class ConcurrentTransition(Exception): class Transition: def __init__( self, - method: Callable[..., Any], + method: Callable[..., str | int | None], source: str | int | Sequence[str | int] | State, - target: str | int | State | None, + target: str | int, on_error: str | int | None, - conditions: list[Callable[[Any], bool]], - permission: str | Callable[[models.Model, UserWithPermissions], bool] | None, + conditions: list[Callable[[_Instance], bool]], + permission: str | Callable[[_Instance, UserWithPermissions], bool] | None, custom: dict[str, _StrOrPromise], ) -> None: self.method = method @@ -99,7 +103,7 @@ def __init__( def name(self) -> str: return self.method.__name__ - def has_perm(self, instance, user: UserWithPermissions) -> bool: + def has_perm(self, instance: _Instance, user: UserWithPermissions) -> bool: if not self.permission: return True if callable(self.permission): @@ -122,7 +126,7 @@ def __eq__(self, other): return False -def get_available_FIELD_transitions(instance, field: FSMFieldMixin) -> Generator[Transition, None, None]: +def get_available_FIELD_transitions(instance: _Instance, field: FSMFieldMixin) -> Generator[Transition, None, None]: """ List of transitions available in current model state with all conditions met @@ -136,7 +140,7 @@ def get_available_FIELD_transitions(instance, field: FSMFieldMixin) -> Generator yield meta.get_transition(curr_state) -def get_all_FIELD_transitions(instance, field: FSMFieldMixin) -> Generator[Transition, None, None]: +def get_all_FIELD_transitions(instance: _Instance, field: FSMFieldMixin) -> Generator[Transition, None, None]: """ List of all transitions available in current model state """ @@ -144,7 +148,7 @@ def get_all_FIELD_transitions(instance, field: FSMFieldMixin) -> Generator[Trans def get_available_user_FIELD_transitions( - instance, user: UserWithPermissions, field: FSMFieldMixin + instance: _Instance, user: UserWithPermissions, field: FSMFieldMixin ) -> Generator[Transition, None, None]: """ List of transitions available in current model state @@ -160,11 +164,11 @@ class FSMMeta: Models methods transitions meta information """ - def __init__(self, field, method) -> None: + def __init__(self, field: FSMFieldMixin, method: Any) -> None: self.field = field - self.transitions: dict[str, Any] = {} # source -> Transition + self.transitions: dict[str, Transition] = {} # source -> Transition - def get_transition(self, source: str): + def get_transition(self, source: str) -> Transition | None: transition = self.transitions.get(source, None) if transition is None: transition = self.transitions.get("*", None) @@ -174,12 +178,12 @@ def get_transition(self, source: str): def add_transition( self, - method: Callable[..., Any], + method: Callable[..., str | int | None], source: str, target: str | int, on_error: str | int | None = None, - conditions: list[Callable[[Any], bool]] = [], - permission: str | Callable[[models.Model, UserWithPermissions], bool] | None = None, + conditions: list[Callable[[_Instance], bool]] = [], + permission: str | Callable[[_Instance, UserWithPermissions], bool] | None = None, custom: dict[str, _StrOrPromise] = {}, ) -> None: if source in self.transitions: @@ -195,7 +199,7 @@ def add_transition( custom=custom, ) - def has_transition(self, state) -> bool: + def has_transition(self, state: str) -> bool: """ Lookup if any transition exists from current model state using current method """ @@ -210,7 +214,7 @@ def has_transition(self, state) -> bool: return False - def conditions_met(self, instance, state) -> bool: + def conditions_met(self, instance: _Instance, state: str) -> bool: """ Check if all conditions have been met """ @@ -224,7 +228,7 @@ def conditions_met(self, instance, state) -> bool: return all(condition(instance) for condition in transition.conditions) - def has_transition_perm(self, instance, state, user: UserWithPermissions) -> bool: + def has_transition_perm(self, instance: _Instance, state: str, user: UserWithPermissions) -> bool: transition = self.get_transition(state) if not transition: @@ -232,7 +236,7 @@ def has_transition_perm(self, instance, state, user: UserWithPermissions) -> boo else: return bool(transition.has_perm(instance, user)) - def next_state(self, current_state): + def next_state(self, current_state: str) -> str | int: transition = self.get_transition(current_state) if transition is None: @@ -240,7 +244,7 @@ def next_state(self, current_state): return transition.target - def exception_state(self, current_state): + def exception_state(self, current_state: str) -> str | int | None: transition = self.get_transition(current_state) if transition is None: @@ -250,15 +254,15 @@ def exception_state(self, current_state): class FSMFieldDescriptor: - def __init__(self, field) -> None: + def __init__(self, field: FSMFieldMixin) -> None: self.field = field - def __get__(self, instance, type=None): + def __get__(self, instance: _Instance, type: Any | None = None) -> Any: if instance is None: return self return self.field.get_state(instance) - def __set__(self, instance, value) -> None: + def __set__(self, instance: _Instance, value: Any) -> None: if self.field.protected and self.field.name in instance.__dict__: raise AttributeError(f"Direct {self.field.name} modification is not allowed") @@ -272,7 +276,7 @@ class FSMFieldMixin(_Field): def __init__(self, *args: Any, **kwargs: Any) -> None: self.protected = kwargs.pop("protected", False) - self.transitions: dict[Any, dict[str, Any]] = {} # cls -> (transitions name -> method) + self.transitions: dict[type[_Model], dict[str, Any]] = {} # cls -> (transitions name -> method) self.state_proxy = {} # state -> ProxyClsRef state_choices = kwargs.pop("state_choices", None) @@ -289,21 +293,21 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) - def deconstruct(self): + def deconstruct(self) -> Any: name, path, args, kwargs = super().deconstruct() if self.protected: kwargs["protected"] = self.protected return name, path, args, kwargs - def get_state(self, instance) -> Any: + def get_state(self, instance: _Instance) -> Any: # The state field may be deferred. We delegate the logic of figuring this out # and loading the deferred field on-demand to Django's built-in DeferredAttribute class. return DeferredAttribute(self).__get__(instance) # type: ignore[attr-defined] - def set_state(self, instance, state: str) -> None: + def set_state(self, instance: _Instance, state: str) -> None: instance.__dict__[self.name] = state - def set_proxy(self, instance, state: str) -> None: + def set_proxy(self, instance: _Instance, state: str) -> None: """ Change class """ @@ -324,7 +328,7 @@ def set_proxy(self, instance, state: str) -> None: instance.__class__ = model - def change_state(self, instance, method, *args: Any, **kwargs: Any): + def change_state(self, instance: _Instance, method: _ToDo, *args: Any, **kwargs: Any) -> Any: meta = method._django_fsm method_name = method.__name__ current_state = self.get_state(instance) @@ -377,7 +381,7 @@ def change_state(self, instance, method, *args: Any, **kwargs: Any): return result - def get_all_transitions(self, instance_cls) -> Generator[Transition, None, None]: + def get_all_transitions(self, instance_cls: type[_Model]) -> Generator[Transition, None, None]: """ Returns [(source, target, name, method)] for all field transitions """ @@ -389,7 +393,7 @@ def get_all_transitions(self, instance_cls) -> Generator[Transition, None, None] for transition in meta.transitions.values(): yield transition - def contribute_to_class(self, cls, name, private_only=False, **kwargs): + def contribute_to_class(self, cls: type[_Model], name: str, private_only: bool = False, **kwargs: Any) -> None: self.base_cls = cls super().contribute_to_class(cls, name, private_only=private_only, **kwargs) @@ -404,7 +408,7 @@ def contribute_to_class(self, cls, name, private_only=False, **kwargs): class_prepared.connect(self._collect_transitions) - def _collect_transitions(self, *args: Any, **kwargs: Any): + def _collect_transitions(self, *args: Any, **kwargs: Any) -> None: sender = kwargs["sender"] if not issubclass(sender, self.base_cls): @@ -456,10 +460,10 @@ class FSMKeyField(FSMFieldMixin, ForeignKey): State Machine support for Django model """ - def get_state(self, instance): + def get_state(self, instance: _Instance) -> _ToDo: return instance.__dict__[self.attname] - def set_state(self, instance, state): + def set_state(self, instance: _Instance, state: str) -> None: instance.__dict__[self.attname] = self.to_python(state) @@ -521,7 +525,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self._update_initial_state() @property - def state_fields(self): + 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): @@ -567,14 +571,14 @@ def save(self, *args: Any, **kwargs: Any) -> None: def transition( - field, + field: FSMFieldMixin, source: str | int | Sequence[str | int] | State = "*", target: str | int | State | None = None, on_error: str | int | None = None, conditions: list[Callable[[Any], bool]] = [], permission: str | Callable[[models.Model, UserWithPermissions], bool] | None = None, custom: dict[str, _StrOrPromise] = {}, -): +) -> _ToDo: """ Method decorator to mark allowed transitions. @@ -596,7 +600,7 @@ def inner_transition(func): func._django_fsm.add_transition(func, source, target, on_error, conditions, permission, custom) @wraps(func) - def _change_state(instance, *args: Any, **kwargs: Any): + def _change_state(instance: _Instance, *args: Any, **kwargs: Any) -> _ToDo: return fsm_meta.field.change_state(instance, func, *args, **kwargs) if not wrapper_installed: @@ -607,7 +611,7 @@ def _change_state(instance, *args: Any, **kwargs: Any): return inner_transition -def can_proceed(bound_method, check_conditions: bool = True) -> bool: +def can_proceed(bound_method: _ToDo, check_conditions: bool = True) -> bool: """ Returns True if model in state allows to call bound_method @@ -624,7 +628,7 @@ def can_proceed(bound_method, 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, user: UserWithPermissions) -> bool: +def has_transition_perm(bound_method: _ToDo, user: UserWithPermissions) -> bool: """ Returns True if model in state allows to call bound_method and user have rights on it """ From 564271a80d8eeab1a9b98c60aee9ebbd6e845fb5 Mon Sep 17 00:00:00 2001 From: Pascal F Date: Wed, 22 Nov 2023 09:02:48 +0100 Subject: [PATCH 07/10] Step 5 --- .mypy.ini | 2 +- django_fsm/__init__.py | 27 +++++++++++++++------------ 2 files changed, 16 insertions(+), 13 deletions(-) diff --git a/.mypy.ini b/.mypy.ini index 75ef771..44ecf4e 100644 --- a/.mypy.ini +++ b/.mypy.ini @@ -20,7 +20,7 @@ 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 +disallow_untyped_defs = True # This one isn't too hard to get passing, but return on investment is lower no_implicit_reexport = True diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index 445c8bd..99409eb 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -49,6 +49,7 @@ IntegerField = models.IntegerField[int, int] ForeignKey = models.ForeignKey[Any, Any] + _StateValue = str | int _Instance = models.Model # TODO: use real type _ToDo = Any # TODO: use real type else: @@ -83,10 +84,10 @@ class ConcurrentTransition(Exception): class Transition: def __init__( self, - method: Callable[..., str | int | None], - source: str | int | Sequence[str | int] | State, - target: str | int, - on_error: str | int | None, + method: Callable[..., _StateValue | Any], + source: _StateValue | Sequence[_StateValue] | State, + target: _StateValue, + on_error: _StateValue | None, conditions: list[Callable[[_Instance], bool]], permission: str | Callable[[_Instance, UserWithPermissions], bool] | None, custom: dict[str, _StrOrPromise], @@ -414,7 +415,7 @@ def _collect_transitions(self, *args: Any, **kwargs: Any) -> None: if not issubclass(sender, self.base_cls): return - def is_field_transition_method(attr): + def is_field_transition_method(attr: _ToDo) -> bool: return ( (inspect.ismethod(attr) or inspect.isfunction(attr)) and hasattr(attr, "_django_fsm") @@ -528,7 +529,7 @@ 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): + def _do_update(self, base_qs, using, pk_val, values, update_fields, forced_update): # type: ignore[no-untyped-def] # _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. @@ -572,13 +573,13 @@ def save(self, *args: Any, **kwargs: Any) -> None: def transition( field: FSMFieldMixin, - source: str | int | Sequence[str | int] | State = "*", + source: str | int | Sequence[str | int] = "*", target: str | int | State | None = None, on_error: str | int | None = None, conditions: list[Callable[[Any], bool]] = [], permission: str | Callable[[models.Model, UserWithPermissions], bool] | None = None, custom: dict[str, _StrOrPromise] = {}, -) -> _ToDo: +) -> Callable[[Any], Any]: """ Method decorator to mark allowed transitions. @@ -586,7 +587,7 @@ def transition( has not changed after the function call. """ - def inner_transition(func): + def inner_transition(func: _ToDo) -> _ToDo: wrapper_installed, fsm_meta = True, getattr(func, "_django_fsm", None) if not fsm_meta: wrapper_installed = False @@ -647,7 +648,7 @@ def has_transition_perm(bound_method: _ToDo, user: UserWithPermissions) -> bool: class State: - def get_state(self, model, transition, result, args=[], kwargs={}): + def get_state(self, model: _Model, transition: Transition, result: Any, args: Any = [], kwargs: Any = {}) -> _ToDo: raise NotImplementedError @@ -655,7 +656,7 @@ class RETURN_VALUE(State): def __init__(self, *allowed_states: Sequence[str | int]) -> None: self.allowed_states = allowed_states if allowed_states else None - def get_state(self, model, transition, result, args=[], kwargs={}): + def get_state(self, model: _Model, transition: Transition, result: Any, args: Any = [], kwargs: Any = {}) -> _ToDo: 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}") @@ -667,7 +668,9 @@ def __init__(self, func: Callable[..., str | int], states: Sequence[str | int] | self.func = func self.allowed_states = states - def get_state(self, model, transition, result, args=[], kwargs={}): + def get_state( + self, model: _Model, transition: Transition, result: _StateValue | Any, args: Any = [], kwargs: Any = {} + ) -> _ToDo: result_state = self.func(model, *args, **kwargs) if self.allowed_states is not None: if result_state not in self.allowed_states: From 93e1248b1a367ee5238f6258d03449914e224241 Mon Sep 17 00:00:00 2001 From: Pascal F Date: Wed, 22 Nov 2023 09:14:34 +0100 Subject: [PATCH 08/10] Step 6 --- django_fsm/__init__.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index 99409eb..33f1c9a 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -45,8 +45,8 @@ _Model = models.Model _Field = models.Field[Any, Any] - CharField = models.CharField[str, str] - IntegerField = models.IntegerField[int, int] + CharField = models.CharField[Any, Any] + IntegerField = models.IntegerField[Any, Any] ForeignKey = models.ForeignKey[Any, Any] _StateValue = str | int @@ -179,10 +179,10 @@ def get_transition(self, source: str) -> Transition | None: def add_transition( self, - method: Callable[..., str | int | None], + method: Callable[..., _StateValue | Any], source: str, - target: str | int, - on_error: str | int | None = None, + target: _StateValue, + on_error: _StateValue | None = None, conditions: list[Callable[[_Instance], bool]] = [], permission: str | Callable[[_Instance, UserWithPermissions], bool] | None = None, custom: dict[str, _StrOrPromise] = {}, @@ -237,7 +237,7 @@ def has_transition_perm(self, instance: _Instance, state: str, user: UserWithPer else: return bool(transition.has_perm(instance, user)) - def next_state(self, current_state: str) -> str | int: + def next_state(self, current_state: str) -> _StateValue: transition = self.get_transition(current_state) if transition is None: @@ -245,7 +245,7 @@ def next_state(self, current_state: str) -> str | int: return transition.target - def exception_state(self, current_state: str) -> str | int | None: + def exception_state(self, current_state: str) -> _StateValue | None: transition = self.get_transition(current_state) if transition is None: @@ -573,9 +573,9 @@ def save(self, *args: Any, **kwargs: Any) -> None: def transition( field: FSMFieldMixin, - source: str | int | Sequence[str | int] = "*", - target: str | int | State | None = None, - on_error: str | int | None = None, + source: _StateValue | Sequence[_StateValue] = "*", + target: _StateValue | State | None = None, + on_error: _StateValue | None = None, conditions: list[Callable[[Any], bool]] = [], permission: str | Callable[[models.Model, UserWithPermissions], bool] | None = None, custom: dict[str, _StrOrPromise] = {}, @@ -653,7 +653,7 @@ def get_state(self, model: _Model, transition: Transition, result: Any, args: An class RETURN_VALUE(State): - def __init__(self, *allowed_states: Sequence[str | int]) -> None: + 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: @@ -664,7 +664,7 @@ def get_state(self, model: _Model, transition: Transition, result: Any, args: An class GET_STATE(State): - def __init__(self, func: Callable[..., str | int], states: Sequence[str | int] | None = None) -> None: + def __init__(self, func: Callable[..., _StateValue | Any], states: Sequence[_StateValue] | None = None) -> None: self.func = func self.allowed_states = states From ec9184b332d480d5e70b7ef2429db32d2216d1f8 Mon Sep 17 00:00:00 2001 From: Pascal F Date: Tue, 28 Nov 2023 07:23:38 +0100 Subject: [PATCH 09/10] Step 7 --- .mypy.ini | 38 ---------- .pre-commit-config.yaml | 3 +- django_fsm/__init__.py | 69 +++++++++++------- pyproject.toml | 73 +++++++++++++++++++ tests/testapp/models.py | 2 +- tests/testapp/tests/test_multidecorators.py | 6 +- .../test_transition_all_except_target.py | 4 +- 7 files changed, 123 insertions(+), 72 deletions(-) delete mode 100644 .mypy.ini diff --git a/.mypy.ini b/.mypy.ini deleted file mode 100644 index 44ecf4e..0000000 --- a/.mypy.ini +++ /dev/null @@ -1,38 +0,0 @@ -[mypy] -# 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 - -[mypy-tests.*] -ignore_errors = True - -[mypy-django_fsm.tests.*] -ignore_errors = True - -[mypy-django_fsm.management.commands.graph_transitions] -ignore_errors = True diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 1bf2d2a..b761d0e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -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 diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index 33f1c9a..85c54ae 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -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,13 +601,14 @@ 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) @@ -601,7 +616,7 @@ def inner_transition(func: _ToDo) -> _ToDo: 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,7 +663,7 @@ 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 @@ -656,7 +671,7 @@ 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: diff --git a/pyproject.toml b/pyproject.toml index 065a619..a5b8e12 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/tests/testapp/models.py b/tests/testapp/models.py index 844a4f4..a6e35e7 100644 --- a/tests/testapp/models.py +++ b/tests/testapp/models.py @@ -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") diff --git a/tests/testapp/tests/test_multidecorators.py b/tests/testapp/tests/test_multidecorators.py index eea9617..697eac3 100644 --- a/tests/testapp/tests/test_multidecorators.py +++ b/tests/testapp/tests/test_multidecorators.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) diff --git a/tests/testapp/tests/test_transition_all_except_target.py b/tests/testapp/tests/test_transition_all_except_target.py index a7765bf..331fa75 100644 --- a/tests/testapp/tests/test_transition_all_except_target.py +++ b/tests/testapp/tests/test_transition_all_except_target.py @@ -8,7 +8,7 @@ from django_fsm import transition -class TestExceptTargetTransitionShortcut(models.Model): +class ExceptTargetTransitionShortcutModel(models.Model): state = FSMField(default="new") @transition(field=state, source="new", target="published") @@ -25,7 +25,7 @@ class Meta: class Test(TestCase): def setUp(self): - self.model = TestExceptTargetTransitionShortcut() + self.model = ExceptTargetTransitionShortcutModel() def test_usecase(self): self.assertEqual(self.model.state, "new") From 3f506d5f5ad6d6dcbc4671f26478983caf806da3 Mon Sep 17 00:00:00 2001 From: Pascal F Date: Wed, 1 May 2024 13:46:21 +0200 Subject: [PATCH 10/10] Step 8 --- .pre-commit-config.yaml | 4 ++-- django_fsm/__init__.py | 36 ++++++++++++++++++------------------ pyproject.toml | 3 ++- 3 files changed, 22 insertions(+), 21 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b761d0e..26bc820 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -50,9 +50,9 @@ repos: - id: ruff - repo: https://github.com/pre-commit/mirrors-mypy - rev: v1.7.1 + rev: v1.10.0 hooks: - id: mypy additional_dependencies: - - django-stubs==4.2.6 + - django-stubs==5.0.0 - django-guardian diff --git a/django_fsm/__init__.py b/django_fsm/__init__.py index 85c54ae..2bf51f9 100644 --- a/django_fsm/__init__.py +++ b/django_fsm/__init__.py @@ -5,11 +5,6 @@ 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 @@ -41,6 +36,11 @@ ] if TYPE_CHECKING: + 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 typing import Self from _typeshed import Incomplete @@ -121,10 +121,10 @@ def has_perm(self, instance: _Instance, user: UserWithPermissions) -> bool: return True return False - def __hash__(self): + def __hash__(self) -> int: return hash(self.name) - def __eq__(self, other): + def __eq__(self, other: object) -> bool: if isinstance(other, str): return other == self.name if isinstance(other, Transition): @@ -240,8 +240,8 @@ def has_transition_perm(self, instance: _Instance, state: str, user: UserWithPer if not transition: return False - else: - return bool(transition.has_perm(instance, user)) + + return bool(transition.has_perm(instance, user)) def next_state(self, current_state: str) -> _StateValue: transition = self.get_transition(current_state) @@ -309,7 +309,7 @@ def deconstruct(self) -> Any: def get_state(self, instance: _Instance) -> Any: # The state field may be deferred. We delegate the logic of figuring this out # and loading the deferred field on-demand to Django's built-in DeferredAttribute class. - return DeferredAttribute(self).__get__(instance) # type: ignore[attr-defined] + return DeferredAttribute(self).__get__(instance) def set_state(self, instance: _Instance, state: str) -> None: instance.__dict__[self.name] = state @@ -479,14 +479,14 @@ class FSMModelMixin(_FSMModel): Mixin that allows refresh_from_db for models with fsm protected fields """ - def _get_protected_fsm_fields(self): - def is_fsm_and_protected(f): + def _get_protected_fsm_fields(self) -> set[str]: + def is_fsm_and_protected(f: object) -> Any: return isinstance(f, FSMFieldMixin) and f.protected - protected_fields = filter(is_fsm_and_protected, self._meta.concrete_fields) + protected_fields: Iterable[Any] = filter(is_fsm_and_protected, self._meta.concrete_fields) # type: ignore[attr-defined, arg-type] return {f.attname for f in protected_fields} - def refresh_from_db(self, *args, **kwargs): + def refresh_from_db(self, *args: Any, **kwargs: Any) -> None: fields = kwargs.pop("fields", None) # Use provided fields, if not set then reload all non-deferred fields.0 @@ -495,7 +495,7 @@ def refresh_from_db(self, *args, **kwargs): protected_fields = self._get_protected_fsm_fields() skipped_fields = deferred_fields.union(protected_fields) - fields = [f.attname for f in self._meta.concrete_fields if f.attname not in skipped_fields] + fields = [f.attname for f in self._meta.concrete_fields if f.attname not in skipped_fields] # type: ignore[attr-defined] kwargs["fields"] = fields super().refresh_from_db(*args, **kwargs) @@ -538,9 +538,9 @@ def state_fields(self) -> Iterable[Any]: def _do_update( self, base_qs: QuerySet[Self], - using: Any, + using: str | None, pk_val: Any, - values: Collection[Any] | None, + values: Collection[tuple[_Field, type[models.Model] | None, Any]], update_fields: Iterable[str] | None, forced_update: bool, ) -> bool: @@ -553,7 +553,7 @@ def _do_update( # 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: bool = super()._do_update( # type: ignore[misc] + updated: bool = super()._do_update( base_qs=base_qs.filter(**state_filter), using=using, pk_val=pk_val, diff --git a/pyproject.toml b/pyproject.toml index a5b8e12..6beb3ed 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,8 +67,9 @@ extend-select = [ "RET", "C", # "B", + "TCH", # trailing comma ] -fixable = ["I"] +fixable = ["I", "TCH"] [tool.ruff.lint.isort]