feat: introduce grace_time_in_seconds parameter to Store to allow a grace period for the store to finish its work before calling cleanup and on_finish

Author: sassanh

.github/workflows/integration_delivery.yml

@@ -16,7 +16,7 @@ jobs:
       - uses: actions/checkout@v4
         name: Checkout
 
-      - name: Save Cached Poetry
+      - name: Load Cached Poetry
         id: cached-poetry
         uses: actions/cache@v4
         with:
@@ -112,36 +112,23 @@ jobs:
           architecture: x64
 
       - name: Load Cached Poetry
-        id: cached-poetry
         uses: actions/cache/restore@v4
+        id: cached-poetry
         with:
           path: |
             ~/.cache
             ~/.local
           key: poetry-${{ hashFiles('poetry.lock') }}
 
-      - name: Test
+      - name: Run Tests
         run: poetry run poe test
 
-      - name: Prepare list of JSON files with mismatching pairs
-        if: failure()
-        run: |
-          mkdir -p artifacts
-          for file in $(find tests/ -name "*.mismatch.json"); do
-            base=${file%.mismatch.json}.json
-            if [[ -f "$base" ]]; then
-              echo "$file" >> artifacts/files_to_upload.txt
-              echo "$base" >> artifacts/files_to_upload.txt
-            fi
-          done
-
-      - name: Collect Mismatching Store Snapshots
-        if: failure()
+      - name: Collect Store Snapshots
         uses: actions/upload-artifact@v4
+        if: always()
         with:
-          name: mismatching-snapshots
-          path: |
-            @artifacts/files_to_upload.txt
+          name: snapshots
+          path: tests/**/results/**/*.jsonc
 
       - name: Collect HTML Coverage Report
         uses: actions/upload-artifact@v4
@@ -163,8 +150,8 @@ jobs:
       - dependencies
     runs-on: ubuntu-latest
     outputs:
-      version: ${{ steps.extract_version.outputs.version }}
-      name: ${{ steps.extract_version.outputs.name }}
+      version: ${{ steps.extract_version.outputs.VERSION }}
+      name: ${{ steps.extract_version.outputs.NAME }}
     steps:
       - uses: actions/checkout@v4
         name: Checkout
@@ -184,14 +171,41 @@ jobs:
             ~/.local
           key: poetry-${{ hashFiles('poetry.lock') }}
 
-      - name: Build
-        run: poetry build
-
       - name: Extract Version
         id: extract_version
         run: |
-          echo "version=$(poetry version --short)" >> "$GITHUB_OUTPUT"
-          echo "name=$(poetry version | cut -d' ' -f1)" >> "$GITHUB_OUTPUT"
+          echo "VERSION=$(poetry version --short)" >> "$GITHUB_OUTPUT"
+          echo "NAME=$(poetry version | cut -d' ' -f1)" >> "$GITHUB_OUTPUT"
+          echo "VERSION=$(poetry version --short)"
+          echo "NAME=$(poetry version | cut -d' ' -f1)"
+
+      - name: Extract Version from CHANGELOG.md
+        id: extract_changelog_version
+        run: |
+          VERSION_CHANGELOG=$(sed -n '3 s/## Version //p' CHANGELOG.md) 
+          echo "VERSION_CHANGELOG=$VERSION_CHANGELOG"
+          if [ "${{ steps.extract_version.outputs.VERSION }}" != "$VERSION_CHANGELOG" ]; then
+            echo "Error: Version extracted from CHANGELOG.md does not match the version in pyproject.toml"
+            exit 1 
+          else
+            echo "Versions are consistent."
+          fi
+
+      - name: Extract Version from Tag
+        if: startsWith(github.ref, 'refs/tags/v')
+        id: extract_tag_version
+        run: |
+          VERSION_TAG=$(sed 's/^v//' <<< ${{ github.ref_name }})
+          echo "VERSION_TAG=$VERSION_TAG"
+          if [ "${{ steps.extract_version.outputs.VERSION }}" != "$VERSION_TAG" ]; then
+            echo "Error: Version extracted from tag does not match the version in pyproject.toml"
+            exit 1 
+          else
+            echo "Versions are consistent."
+          fi
+
+      - name: Build
+        run: poetry build
 
       - name: Upload wheel
         uses: actions/upload-artifact@v4
@@ -240,19 +254,20 @@ jobs:
           verbose: true
 
   release:
+    if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/v')
     name: Release
     needs:
       - type-check
       - lint
       - test
       - build
       - pypi-publish
+    runs-on: ubuntu-latest
     environment:
       name: release
-    runs-on: ubuntu-latest
+      url: https://pypi.org/p/${{ needs.build.outputs.name }}
     permissions:
       contents: write
-    if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/v')
     steps:
       - name: Procure Wheel
         uses: actions/download-artifact@v4

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## Version 0.14.1
+
+- feat: introduce `grace_time_in_seconds` parameter to `Store` to allow a grace
+  period for the store to finish its work before calling `cleanup` and `on_finish`
+
 ## Version 0.14.0
 
 - refactor: `Store` no longer aggregates changes, it now calls listeners with every

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "python-redux"
-version = "0.14.0"
+version = "0.14.1"
 description = "Redux implementation for Python"
 authors = ["Sassan Haradji <[email protected]>"]
 license = "Apache-2.0"
@@ -16,8 +16,8 @@ optional = true
 
 [tool.poetry.group.dev.dependencies]
 poethepoet = "^0.24.4"
-pyright = "^1.1.354"
-ruff = "^0.3.3"
+pyright = "^1.1.357"
+ruff = "^0.3.5"
 pytest = "^8.1.1"
 pytest-cov = "^4.1.0"
 pytest-timeout = "^2.3.1"

pyproject.toml

@@ -42,8 +42,8 @@ class BaseEvent(Immutable): ...
 
 class CompleteReducerResult(Immutable, Generic[State, Action, Event]):
     state: State
-    actions: list[Action] | None = None
-    events: list[Event] | None = None
+    actions: Sequence[Action] | None = None
+    events: Sequence[Event] | None = None
 
 
 ReducerResult = CompleteReducerResult[State, Action, Event] | State
@@ -112,6 +112,7 @@ class CreateStoreOptions(Immutable, Generic[Action, Event]):
     event_middlewares: Sequence[EventMiddleware[Event]] = field(default_factory=list)
     task_creator: TaskCreator | None = None
     on_finish: Callable[[], Any] | None = None
+    grace_time_in_seconds: float = 1
 
 
 class AutorunOptions(Immutable, Generic[AutorunOriginalReturnType]):

redux/basic_types.py

@@ -7,7 +7,7 @@
 import queue
 import weakref
 from collections import defaultdict
-from threading import Lock
+from threading import Lock, Thread
 from typing import Any, Callable, Generic, cast
 
 from redux.autorun import Autorun
@@ -50,7 +50,6 @@ def __init__(
         options: CreateStoreOptions[Action, Event] | None = None,
     ) -> None:
         """Create a new store."""
-        self.finished = False
         self.store_options = options or CreateStoreOptions()
         self.reducer = reducer
         self._create_task = self.store_options.task_creator
@@ -125,8 +124,8 @@ def _run_actions(self: Store[State, Action, Event]) -> None:
 
     def _run_event_handlers(self: Store[State, Action, Event]) -> None:
         event = self._events.pop(0)
-        for event_handler_ in self._event_handlers[type(event)].copy():
-            self._event_handlers_queue.put_nowait((event_handler_, event))
+        for event_handler in self._event_handlers[type(event)].copy():
+            self._event_handlers_queue.put_nowait((event_handler, event))
 
     def run(self: Store[State, Action, Event]) -> None:
         """Run the store."""
@@ -137,13 +136,6 @@ def run(self: Store[State, Action, Event]) -> None:
 
                 if len(self._events) > 0:
                     self._run_event_handlers()
-        if (
-            self.finished
-            and self._actions == []
-            and self._events == []
-            and not any(worker.is_alive() for worker in self._workers)
-        ):
-            self.clean_up()
 
     def clean_up(self: Store[State, Action, Event]) -> None:
         """Clean up the store."""
@@ -152,8 +144,6 @@ def clean_up(self: Store[State, Action, Event]) -> None:
         self._workers.clear()
         self._listeners.clear()
         self._event_handlers.clear()
-        if self.store_options.on_finish:
-            self.store_options.on_finish()
 
     def dispatch(
         self: Store[State, Action, Event],
@@ -225,10 +215,30 @@ def unsubscribe() -> None:
 
         return unsubscribe
 
+    def wait_for_store_to_finish(self: Store[State, Action, Event]) -> None:
+        """Wait for the store to finish."""
+        import time
+
+        while True:
+            if (
+                self._actions == []
+                and self._events == []
+                and self._event_handlers_queue.qsize() == 0
+            ):
+                time.sleep(self.store_options.grace_time_in_seconds)
+                self._event_handlers_queue.join()
+                for _ in range(self.store_options.threads):
+                    self._event_handlers_queue.put_nowait(None)
+                self._event_handlers_queue.join()
+                self.clean_up()
+                if self.store_options.on_finish:
+                    self.store_options.on_finish()
+                break
+            time.sleep(0.1)
+
     def _handle_finish_event(self: Store[State, Action, Event]) -> None:
-        for _ in range(self.store_options.threads):
-            self._event_handlers_queue.put_nowait(None)
-        self.finished = True
+        thread = Thread(target=self.wait_for_store_to_finish)
+        thread.start()
 
     def autorun(
         self: Store[State, Action, Event],

redux/main.py

@@ -39,7 +39,7 @@ def __init__(
         self.results_dir = path.parent / 'results' / file / test_id.split('::')[-1][5:]
         if self.results_dir.exists():
             for file in self.results_dir.glob(
-                'store-*' if override else 'store-*.mismatch.json',
+                'store-*.jsonc' if override else 'store-*.mismatch.jsonc',
             ):
                 file.unlink()  # pragma: no cover
         self.results_dir.mkdir(parents=True, exist_ok=True)