chore: New example of model based on #358 (#363)

Author: fgmacedo

pyproject.toml

@@ -56,7 +56,7 @@ requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.pytest.ini_options]
-addopts = "--ignore-glob=docs/conf.py --ignore-glob=statemachine/factory_* --ignore-glob=docs/auto_examples/* --ignore-glob=docs/_build/* --cov --cov-config .coveragerc --doctest-glob='*.md' --doctest-modules --doctest-continue-on-failure"
+addopts = "--ignore=docs/conf.py --ignore=docs/auto_examples/ --ignore=docs/_build/ --ignore=tests/examples/ --cov --cov-config .coveragerc --doctest-glob='*.md' --doctest-modules --doctest-continue-on-failure"
 doctest_optionflags = "ELLIPSIS IGNORE_EXCEPTION_DETAIL NORMALIZE_WHITESPACE IGNORE_EXCEPTION_DETAIL"
 
 [tool.mypy]

tests/examples/persistent_model_machine.py

@@ -0,0 +1,144 @@
+"""
+Persistent domain model
+=======================
+
+An example originated from a question: "How to save state to disk?". There are many ways to
+implement this, but you can get an insight of one possibility. This example implements a custom
+domain model that persists it's state using a generic strategy that can be extended to any storage
+format.
+
+Original `issue <https://github.com/fgmacedo/python-statemachine/issues/358>`_.
+
+
+Resource management state machine
+---------------------------------
+
+Given a simple on/off machine for resource management.
+
+"""
+
+import tempfile
+from abc import ABC
+from abc import abstractmethod
+
+from statemachine import State
+from statemachine import StateMachine
+
+
+class ResourceManagement(StateMachine):
+    power_off = State(initial=True)
+    power_on = State()
+
+    turn_on = power_off.to(power_on)
+    shutdown = power_on.to(power_off)
+
+
+# %%
+# Abstract model with persistency protocol
+# ----------------------------------------
+#
+# Abstract Base Class for persistent models.
+# Subclasses should implement concrete strategies for:
+#
+# - `_read_state`: Read the state from the concrete persistent layer.
+# - `_write_state`: Write the state from the concrete persistent layer.
+
+
+class AbstractPersistentModel(ABC):
+    """Abstract Base Class for persistent models.
+
+    Subclasses should implement concrete strategies for:
+
+    - `_read_state`: Read the state from the concrete persistent layer.
+    - `_write_state`: Write the state from the concrete persistent layer.
+    """
+
+    def __init__(self):
+        self._state = None
+
+    def __repr__(self):
+        return f"{type(self).__name__}(state={self.state})"
+
+    @property
+    def state(self):
+        if self._state is None:
+            self._state = self._read_state()
+        return self._state
+
+    @state.setter
+    def state(self, value):
+        self._state = value
+        self._write_state(value)
+
+    @abstractmethod
+    def _read_state(self):
+        ...
+
+    @abstractmethod
+    def _write_state(self, value):
+        ...
+
+
+# %%
+# FilePersistentModel: Concrete model strategy
+# -----------------------
+#
+# A concrete implementation of the generic storage protocol above, that reads and writes to a file
+# in plain text.
+
+
+class FilePersistentModel(AbstractPersistentModel):
+    """A concrete implementation of a storage strategy for a Model
+    that reads and writes to a file.
+    """
+
+    def __init__(self, file):
+        super().__init__()
+        self.file = file
+
+    def _read_state(self):
+        self.file.seek(0)
+        state = self.file.read().strip()
+        return state if state != "" else None
+
+    def _write_state(self, value):
+        self.file.seek(0)
+        self.file.truncate(0)
+        self.file.write(value)
+
+
+# %%
+# Given a temporary file to store our state.
+
+state_file = tempfile.TemporaryFile(mode="r+")
+
+# %%
+# Let's create instances and test the persistence.
+
+model = FilePersistentModel(file=state_file)
+sm = ResourceManagement(model=model)
+
+print(f"Initial state: {sm.current_state.id}")
+
+sm.send("turn_on")
+
+print(f"State after transition: {sm.current_state.id}")
+
+# %%
+# Remove the instances from memory.
+
+del sm
+del model
+
+# %%
+# Restore the previous state from disk.
+
+model = FilePersistentModel(file=state_file)
+sm = ResourceManagement(model=model)
+
+print(f"State restored from file system: {sm.current_state.id}")
+
+# %%
+# Closing the file (the temporary file will be removed).
+
+state_file.close()

tests/helpers.py

@@ -0,0 +1,10 @@
+import importlib
+from pathlib import Path
+
+
+def import_module_by_path(src_file: Path):
+    module_name = str(src_file).replace("/", ".")
+    try:
+        return importlib.import_module(module_name)
+    except ModuleNotFoundError:
+        return

tests/scrape_images.py

@@ -1,11 +1,12 @@
-import importlib
 import re
 
 from sphinx_gallery.scrapers import figure_rst
 
 from statemachine.contrib.diagram import DotGraphMachine
 from statemachine.factory import StateMachineMetaclass
 
+from .helpers import import_module_by_path
+
 
 class MachineScraper:
     """Scrapes images of the statemachines defined into the examples for the gallery"""
@@ -25,11 +26,7 @@ def _get_module(self, src_file):
         if len(module_name) != 1:
             return
 
-        module_name = module_name[0].replace("/", ".")
-        try:
-            return importlib.import_module(module_name)
-        except ModuleNotFoundError:
-            return
+        return import_module_by_path(module_name[0])
 
     def generate_image(self, sm_class, original_path):
         image_path = self.re_replace_png_extension.sub(".svg", original_path)

tests/test_examples.py

@@ -0,0 +1,29 @@
+from pathlib import Path
+
+import pytest
+
+from .helpers import import_module_by_path
+
+
+def pytest_generate_tests(metafunc):
+    if "example_file_wrapper" not in metafunc.fixturenames:
+        return
+
+    file_names = [
+        pytest.param(example_path, id=f"{example_path}")
+        for example_path in Path("tests/examples").glob("**/*_machine.py")
+    ]
+    metafunc.parametrize("file_name", file_names)
+
+
+@pytest.fixture()
+def example_file_wrapper(file_name):
+    def execute_file_wrapper():
+        import_module_by_path(file_name)
+
+    return execute_file_wrapper
+
+
+def test_example(example_file_wrapper):
+    """Import the example file so the module is executed"""
+    example_file_wrapper()