URL: https://linuxfr.org/users/linkdd/journaux/sunday-python-pattern-une-machine-a-etat-toute-simple Title: Sunday Python Pattern : Une machine à état toute simple Authors: David Delassus Date: 2022年04月17日T03:14:11+02:00 License: CC By-SA Tags: design, pattern, python et machine Score: 29 Bonjour Nal, Il y a un "design pattern" que je réutilise souvent dans différent langages pour découper la logique métier en plusieurs petit bout de code bien séparés et facilement testable. Ce design pattern s'apparente très fortement à une machine à état : - on a la machine a état qui possède un contexte (des données qui seront manipulées et modifiées par l'exécution de l'algorithme) - chaque état va agir sur ce contexte et retourner l'état suivant à exécuter, ou `None` si on est arrivé au bout de l'algorithme Ici, les transitions d'un état vers un autre sont modelées par le code : l'état actuel retourne l'état suivant Pour les tests c'est assez simple : étant donné un contexte, quel est le type de retour de mon état ? ## Une implémentation toute simple Allez, un peu de code Python pour imager la chose... Au passage, j'en ai fait une librairie disponible sur [Github](https://github.com/linkdd/easy_fsm). Tout d'abord, on va définir le contexte : ```python from typing import TypeVar Context = TypeVar("Context") ``` Le contexte peut être tout et n'importe quoi, mais on va explicitement le nommer pour la suite (comme ça, des outils tels que **[mypy](http://mypy-lang.org/)** vont pouvoir vérifier que l'on fait correctement les choses). Ensuite, on va définir ce qu'est un état : ```python from typing import Protocol, Generic, Optional class State(Protocol, Generic[Context]): def run(self, context: Context) -> Optional["State[Context]"] raise NotImplementedError ``` **NB :** Les guillemets sur le type de retour sont nécessaires, car `State` n'est pas encore complètement défini à cet endroit (cela ne pose aucun problème à mypy). Ici, c'est l'équivalent Python d'une interface générique, en TypeScript cela donnerait : ```typescript interface State { run(context: Context): State | null } ``` C'est ce que l'on va utiliser par la suite pour implémenter les différentes étapes de notre algorithme. Enfin, on peut définir notre machine à état : ```python class StateMachine(Generic[Context]): def __init__(self, context: Context): self.context = context def run_from(self, state: Optional[State[Context]]): while state is not None: state = state.run(self.context) ``` Et c'est tout. Tant que l'on a un état a exécuter, on continu. C'est l'état qui contrôle le déroulement de l'algorithme. ## Un premier exemple simple Prenons un exemple tout simple, calculer une [suite de syracuse](https://fr.wikipedia.org/wiki/Conjecture_de_Syracuse) : ```python from dataclasses import dataclass, field @dataclass class Stats: altitude: int = 0 fly_time: int = 0 suite: list[int] = field(default_factory=list) class ComputeNextNumber(State[Stats]): def __init__(self, n: int): self.n = n def run(self, context: Stats) -> Optional[State[Stats]]: context.altitude = max(context.altitude, self.n) context.fly_time += 1 context.suite.append(self.n) if self.n == 1: return None elif self.n % 2 == 0: return ComputeNextNumber(self.n // 2) else: return ComputeNextNumber(3 * self.n + 1) class Syracuse(StateMachine[Stats]): def __init__(self): super().__init__(Stats()) def compute(self, n: int) -> None: self.run_from(ComputeNextNumber(n)) fsm = Syracuse() fsm.compute(5) assert fsm.context.altitude == 16 assert fsm.context.fly_time == 6 assert fsm.context.suite == [5, 16, 8, 4, 2, 1] ``` L'entièreté de l'algorithme réside dans la classe `ComputeNextNumber`. Dans cet exemple, le type associé à `Context` est `Stats` (grâce à `State[Stats]` et `StateMachine[Stats]`), et mypy sera en mesure de le vérifier. ## Un exemple plus concret Le second exemple représente un cas plus concret, si on essayait de parser une requête HTTP ? Pour info, une requête HTTP est composée de : - une ligne de requête (version du protocole, verbe HTTP, URL) - zéro ou plusieurs en-têtes - une ligne vide - le corps de la requête Par exemple : ``` POST /hello HTTP/1.1 Host: example.com world ``` **NB :** Dans les exemples de code suivant, je vais omettre le détail de l'implémentation pour se focaliser sur l'utilisation de ce design pattern. Je vois ici 3 étapes : 1. lire la ligne de requête 2. lire une ligne d'en-tête 3. lire le corps de la requête En vérité, il y a une 4è étape qui peut intervenir à plusieurs endroits, pour la gestion d'erreur. Lançons nous dans le code : ```python class HTTPContext: # ... class ParseRequestLine(State[HTTPContext]): def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]: line = context.socket.readline() if not line: # EOF return UnexpectedError(RuntimeError("EOF")) try: verb, url, protocol = self._parse_request_line(line) return ParseHeaderLine(verb, url, protocol) except Exception as err: return UnexpectedError(err) def _parse_request_line(self, line: str) -> tuple[str, str, str]: # ... class ParseHeaderLine(State[HTTPContext]): def __init__(self, verb: str, url: str, protocol: str, headers=None): if headers is None: headers = {} self.verb = verb self.url = url self.protocol = protocol self.headers = headers def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]: line = context.socket.readline() if not line: # EOF return UnexpectedError(RuntimeError("EOF")) line = line.strip() # delete \r\n if not line: return ReadBody(self.verb, self.url, self.protocol, self.headers) try: header_name, header_val = self._parse_header_line(line) except Exception as err: return UnexpectedError(err) self.headers[header_name] = header_val return self # MaximumDepthRecursion avoided!!!! def _parse_header_line(self, line: str) -> tuple[str, str]: # ... class ReadBody(State[HTTPContext]): def __init__(self, verb: str, url: str, protocol: str, headers: dict[str, str]): self.verb = verb self.url = url self.protocol = protocol self.headers = headers def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]: body = context.socket.read() context.set_request_info( verb=self.verb, url=self.url, protocol=self.protocol, headers=self.headers, body=self.body, ) return None class UnexpectedError(State[HTTPContext]): def __init__(self, err: Exception): self.err = err def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]: context.set_error_info(self.err) return None ``` Dans cet exemple, on voit tout de suite les avantages : - on a du code récursif sans limite de récursion, en effet, on quitte la fonction avant de re-rentrer dedans, donc la pile d'appel n'est pas surchargée - chaque étape de l'algorithme est bien séparée, on évite ainsi une fonction gigantesque - une gestion d'erreur plutôt simple et sans `goto`, en effet il suffit de retourner l'état qui va traiter l'erreur - facile à tester, on va pouvoir "mocker" le `HTTPContext` et vérifier le type de retour de la méthode `run()` de chaque étape ## Conclusion Permettre à une fonction de retourner ce que l'appellant doit exécuter ensuite est un pattern très pratique, surtout quand il est question de changement de spec. Imaginons que la syntaxe de notre requête HTTP change, ici seul le code métier devra être revu, et en aucun cas le code qui l'appelle. Le code qui appelle notre algorithme est entièrement piloté par notre algorithme au final (c'est nous qui lui disons quoi faire ensuite). J'ai simplifié pas mal de code métier dans différent projet en utilisant ce concept, que cela soit en Elixir, en Go, en Python, ou en TypeScript (je serais bien curieux de voir ce que cela donne en Rust). Et toi Nal ? Tu en as des pattern sympa comme ça ? - implémentation Python : [easy_fsm](https://github.com/linkdd/easy_fsm)

AltStyle によって変換されたページ (->オリジナル) /