Expérimentation Viewflow FSM avec instruction

[ddahan]

Cette issue a pour but de partager une expérimentation, pouvant servir de base à une future réflexion sur la modélisation de l'instruction, avec son historique lié.

• C'est un "POC" technique, rien n'est complet, d'où l'idée que ce ne soit pas une PR.
• Ne pas faire attention pour le moment si des états ne semblent pas être les bons, ou avec la logique métier qui est "bidon"

Le code actuel

• en vrac, mais commenté
• entièrement auto-suffisant, ne s'appuie (presque) pas sur un existant

from __future__ import annotations

from dataclasses import dataclass

from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.utils import dateformat, formats

from viewflow import fsm

from data.behaviours import AutoValidable, TimeStampable

User = get_user_model()


class InstructorRole(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, related_name="instructor_roles", on_delete=models.PROTECT)


class VisorRole(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, related_name="visor_roles", on_delete=models.PROTECT)


class SignerRole(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, related_name="signer_roles", on_delete=models.PROTECT)


# NOTE: on pourrait aussi avoir un unique rôle `DeclarationProcessorRole` avec can_instruct, can_vise et can_sign


class DeclarationState(models.TextChoices):
    DRAFT = "DRAFT", "brouillon"
    IN_POOL = "IN_POOL", "dans le pool des déclarations à instruire"
    DISPATCHED = "DISPATCHED", "affectée à un instructeur"
    UNDER_INSTRUCTION = "UNDER_INSTRUCTION", "en cours d'instruction"
    AWAITING_PRODUCER = "AWAITING_PRODUCER", "en attente d'un retour de la part du producteur"

    # États de validation hiérarchique
    IN_VISA_POOL = "IN_VISA_POOL", "en attente de visa (niveau 2)"
    VISA_IN_PROGRESS = "VISA_IN_PROGRESS", "en cours de validation par le viseur"
    IN_SIGNATURE_POOL = "IN_SIGNATURE_POOL", "en attente de signature (niveau 3)"
    SIGNATURE_IN_PROGRESS = "SIGNATURE_IN_PROGRESS", "en cours de validation par le signataire"

    # États finaux
    APPROVED = "APPROVED", "approuvée"
    REJECTED = "REJECTED", "rejetée"
    TIMED_OUT = "TIMED_OUT", "délai imparti dépassé"
    CANCELLED = "CANCELLED", "annulée par le producteur"

    # TODO: ajouter les conditions/permissions (ex : instructeur doit avoir le rôle)


FINAL_STATES = [
    DeclarationState.APPROVED,
    DeclarationState.REJECTED,
    DeclarationState.TIMED_OUT,
    DeclarationState.CANCELLED,
]


class RejectionReason(models.TextChoices):
    MISSING_DATA = "MISSING_DATA", "Le dossier manque des données nécessaires"
    MEDICINE = "MEDICINE", "Le complément répond à la définition du médicament"
    INCOMPATIBLE_RECOMMENDATIONS = "INCOMPATIBLE_RECOMMENDATIONS", "Recommandations d'emploi incompatibles"


class DeclarationFlow:
    state = fsm.State(DeclarationState)

    #####################################################################################
    # méthodes attendues par Viewflow
    #####################################################################################

    def __init__(self, declaration):
        self.declaration = declaration

    @state.setter()
    def _set_declaration_state(self, value):
        self.declaration.state = value

    @state.getter()
    def _get_declaration_state(self):
        return self.declaration.state

    #####################################################################################
    # helpers
    #####################################################################################

    def create_log(self, description, changed_by=None) -> DeclarationChangeLog:
        """Wrapper pour éviter des répétitions"""
        return DeclarationChangeLog.objects.create(
            declaration=self.declaration,
            new_state=self.declaration.state,
            description=description,
            changed_by=changed_by,
        )

    #####################################################################################
    # méthodes de transitions contenant la logique métier
    #####################################################################################

    @state.transition(source=DeclarationState.DRAFT, target=DeclarationState.IN_POOL)
    @transaction.atomic()
    def submit(self):
        # FIXME: logique métier ici
        self.create_log(
            description="La déclaration a été soumise pour instruction.",
            changed_by=self.declaration.author,
        )

    @state.transition(source=DeclarationState.IN_POOL, target=DeclarationState.DISPATCHED)
    @transaction.atomic()
    def affect_to_instructor(self):
        self.declaration.instructor = InstructorRole.objects.last().user  # FIXME: (vraie) logique métier ici
        self.declaration.save()
        self.create_log(
            description=f"Le système a affecté la déclaration à {self.declaration.instructor.name} pour instruction.",
        )

    @state.transition(source=DeclarationState.DISPATCHED, target=DeclarationState.UNDER_INSTRUCTION)
    @transaction.atomic()
    def start_instruction(self):
        # FIXME: logique métier ici
        self.create_log(
            description="L'instruction de la déclaration a démarré.",
            changed_by=self.declaration.instructor,
        )

    @state.transition(
        source={DeclarationState.UNDER_INSTRUCTION, DeclarationState.SIGNATURE_IN_PROGRESS},
        target=DeclarationState.REJECTED,
    )
    @transaction.atomic()
    def reject(self, reason: RejectionReason, author: User):
        # FIXME: logique métier ici
        self.rejection_reason = reason
        self.declaration.save()
        self.create_log(
            description=f"La déclaration à été rejetée par {author.name} pour le motif suivant : {reason.value}"
        )


@dataclass
class ConsistencyRule:
    err_msg: str
    condition: bool


class Declaration(AutoValidable, models.Model):
    state = models.CharField(choices=DeclarationState.choices, default=DeclarationState.DRAFT)
    author = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="authored_declarations"
    )
    instructor = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        blank=True,
        null=True,
        related_name="instructed_declarations",
    )
    # autres champs

    def clean(self):
        """Vérifie les règles de consistence de l'objet avant toute sauvegarde.
        TODO: essayer de voir si on peut pas lier ces règles aux transitions, pour ne pas éparpiller la logique.
        """
        rules = [
            ConsistencyRule(
                f"L'instructeur spécifié (ou non) n'est pas compatible avec l'état «{self.get_state_display()}»",
                bool(self.instructor)
                == (self.state not in [DeclarationState.DRAFT, DeclarationState.IN_POOL] + FINAL_STATES),
            ),
            # FIXME: ajouter les autres règles ici
        ]

        for rule in rules:
            if not rule.condition:
                raise ValidationError(rule.err_msg)

    @transaction.atomic()
    def save(self, *args, **kwargs):
        """Log supplémentaire pour avoir la création de l'objet visible directement dans l'historique (car aucune transition n'est créée à la création de l'objet)"""
        if self._state.adding:
            super().save(*args, **kwargs)
            DeclarationChangeLog.objects.create(
                declaration=self,
                new_state=self.state,
                description="La déclaration a été créée en mode brouillon.",
                changed_by=self.author,
            )
        else:
            super().save(*args, **kwargs)


def formatted_date(date) -> str:
    """Utilise un outil Django pour formater un datetime proprement."""
    # TODO: déplacer dans les utils

    date_format = formats.get_format("SHORT_DATETIME_FORMAT")
    return dateformat.format(date, date_format)


class DeclarationChangeLog(TimeStampable, models.Model):
    """Objet créé à chaque changement d'état de la déclaration"""

    class Meta:
        ordering = ("-creation_date",)

    declaration = models.ForeignKey(Declaration, related_name="change_logs", on_delete=models.CASCADE)
    new_state = models.CharField(
        "nouvel état", choices=DeclarationState
    )  # l'ancien est déduit à partir du log précédent
    description = models.TextField()
    changed_by = models.ForeignKey(settings.AUTH_USER_MODEL, blank=True, null=True, on_delete=models.CASCADE)

    @property
    def previous_log(self) -> DeclarationChangeLog | None:
        return (
            self.declaration.change_logs.filter(creation_date__lt=self.creation_date)
            .order_by("-creation_date")
            .first()
        )

    @property
    def previous_state(self) -> str | None:
        return self.previous_log.new_state if self.previous_log else None

    def __str__(self):
        return f"[#{self.declaration.id}] {formatted_date(self.creation_date)} : {self.previous_state or '∅'} → {self.new_state}"

Exemple d'utilisation

In [1]: from pocfsm.models import DeclarationFlow
In [2]: decla = Declaration.objects.create(author=User.objects.last())
In [3]: flow = DeclarationFlow(decla)

In [4]: flow.submit()
In [5]: flow.affect_to_instructor()
In [6]: flow.start_instruction()

Si on essaie de retirer l'instructeur :

[alemangui]

Merci pour ce POC ! On prendra certainement graduellement des éléments dessus. Je le ferme pour l'instant pour ne laisser que les bugs/feature-requests mais nous le gardons sous le coud pour la suite