Improve process future tasks command

.github/workflows/publish.yml

@@ -8,22 +8,22 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v2
-
+    - uses: actions/checkout@v4
+      with:
+        persist-credentials: false
     - name: Set up Python
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
-        python-version: '3.9'
-        architecture: 'x64'
+        python-version: "3.x"
 
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install setuptools wheel twine
+        pip install build twine
 
     - name: Build source and binary distribution package
       run: |
-        python setup.py sdist bdist_wheel
+        python -m build
       env:
         PACKAGE_VERSION: ${{ github.ref }}
 

.github/workflows/test.yml

@@ -8,23 +8,17 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-          - '3.9'
           - '3.10'
           - '3.11'
           - '3.12'
           - '3.13'
         django-version:
           - '4.2'
-          - '5.0'
           - '5.1'
-        exclude:
-          - python-version: '3.9'
-            django-version: '5.0'
-          - python-version: '3.9'
-            django-version: '5.1'
+          - '5.2'
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v2
@@ -34,23 +28,16 @@ jobs:
       - name: Install dependencies and package
         run: |
           python -m pip install --upgrade pip
-          pip install -r requirements.txt
+          pip install -e ".[dev]"
           pip install django~=${{ matrix.django-version }}.0
 
       - name: Run lint and code review
         run: |
           pre-commit run --all-files
       - name: Run tests with coverage
         run: |
-          # prepare Django project: link all necessary data from the test project into the root directory
-          # Hint: Simply changing the directory does not work (leads to missing files in coverage report)
-          ln -s ./tests/core core
-          ln -s ./tests/testapp testapp
-          ln -s ./tests/manage.py manage.py
           # run tests with coverage
-          coverage run \
-            --source='./django_future_tasks' \
-            manage.py test
+          pytest --cov=django_future_tasks tests
           coverage xml
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v3

.gitignore

@@ -4,3 +4,6 @@ __pycache__
 db.sqlite3
 build/*
 tests/static/*
+.python-version
+.coverage
+uv.lock

.pre-commit-config.yaml

@@ -20,8 +20,8 @@ repos:
       - id: add-trailing-comma
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.9
+    rev: v0.12.8
     hooks:
-      - id: ruff
+      - id: ruff-check
         args: [ --fix ]
       - id: ruff-format

CHANGELOG.md

@@ -5,6 +5,43 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Use `timezone.now()` instead of `datetime.now()` in `PeriodicFutureTask.save()`.
+- Use `condition` instead of `check` attribute in `CheckConstraint` with `django>=5.1`.
+
+### Changed
+
+- Improve `process_future_tasks` command.
+  - Terminate after processing the current task instead of the current task batch if SIGINT/SIGTERM is received.
+  - Remove unnecessary waiting for new tasks when there are already tasks that can be processed.
+  - Configurable waiting duration for new tasks.
+
+### Added
+
+- Support for Python 3.14.
+- Support for Django 5.2.
+
+### Removed
+
+- Support for Python 3.9.
+- Support for Django 5.0.
+
+## [1.3.2]
+
+### Changed
+
+- Migrate to `pytest`.
+- Migrate to `pyproject.toml`.
+
+## [1.3.1]
+
+### Fixed
+
+- Fix infinite loop in populate_periodic_future_tasks on IntegrityError
+
 ## [1.3.0]
 
 ### Added
@@ -57,7 +94,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Initial setup.
 
-[Unreleased]: https://github.com/anexia/django-future-tasks/compare/1.3.0...HEAD
+[Unreleased]: https://github.com/anexia/django-future-tasks/compare/1.3.2...HEAD
+[1.3.2]: https://github.com/anexia/django-future-tasks/releases/tag/1.3.2
+[1.3.1]: https://github.com/anexia/django-future-tasks/releases/tag/1.3.1
 [1.3.0]: https://github.com/anexia/django-future-tasks/releases/tag/1.3.0
 [1.2.1]: https://github.com/anexia/django-future-tasks/releases/tag/1.2.1
 [1.2.0]: https://github.com/anexia/django-future-tasks/releases/tag/1.2.0

README.md

@@ -63,9 +63,10 @@ python manage.py populate_periodic_future_tasks
 
 If your project uses an older version of Django or Django Rest Framework, you can choose an older version of this project.
 
-| This Project | Python Version              | Django Version |
-|--------------|-----------------------------|----------------|
-| 1.3.*        | 3.9, 3.10, 3.11, 3.12, 3.13 | 4.2, 5.0, 5.1  |
-| 1.2.*        | 3.8, 3.9, 3.10, 3.11        | 3.2, 4.1, 4.2  |
-| 1.1.*        | 3.8, 3.9, 3.10, 3.11        | 3.2, 4.1, 4.2  |
-| 1.0.*        | 3.8, 3.9, 3.10, 3.11        | 3.2, 4.0, 4.1  |
+| This Project | Python Version               | Django Version |
+|--------------|------------------------------|----------------|
+| 1.4.*        | 3.10, 3.11, 3.12, 3.13, 3.14 | 4.2, 5.1, 5.2  |
+| 1.3.*        | 3.9, 3.10, 3.11, 3.12, 3.13  | 4.2, 5.0, 5.1  |
+| 1.2.*        | 3.8, 3.9, 3.10, 3.11         | 3.2, 4.1, 4.2  |
+| 1.1.*        | 3.8, 3.9, 3.10, 3.11         | 3.2, 4.1, 4.2  |
+| 1.0.*        | 3.8, 3.9, 3.10, 3.11         | 3.2, 4.0, 4.1  |

django_future_tasks/management/commands/populate_periodic_future_tasks.py

@@ -55,8 +55,7 @@ def handle_tick(self):
             for dt in relevant_dts:
                 if (
                     p_task.max_number_of_executions is not None
-                    and self.number_of_corresponding_single_tasks(p_task)
-                    >= p_task.max_number_of_executions
+                    and self.number_of_corresponding_single_tasks(p_task) >= p_task.max_number_of_executions
                 ) or (p_task.end_time is not None and p_task.end_time < dt):
                     p_task.is_active = False
                     break

django_future_tasks/management/commands/process_future_tasks.py

@@ -21,101 +21,103 @@ class Command(BaseCommand):
 
     current_task_pk = None
 
-    def add_arguments(self, parser):
-        parser.add_argument(
-            "--onetimerun",
-            action="append",
-            type=bool,
-            default=False,
-            help="Run command only one times",
-        )
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        # The command will run as long as the `_running` attribute is
+        # set to `True`. To safely quit the command, just set this attribute to `False` and the
+        # command will finish a running tick and quit afterwards.
+        self._running = True
+
+        # Register system signal handler to gracefully quit the service when
+        # getting a `SIGINT` or `SIGTERM` signal (e.g. by CTRL+C).
+        signal.signal(signal.SIGINT, self._handle_termination)
+        signal.signal(signal.SIGTERM, self._handle_termination)
 
     def _handle_termination(self, *args, **kwargs):
+        # Mark the task as interrupted in case the command will receive a SIGKILL before the task was completed.
+        # If the command terminates graciously instead, the task will be finished and marked as done again by the
+        # main loop.
         try:
             current_task = FutureTask.objects.get(pk=self.current_task_pk)
             current_task.status = FutureTask.FUTURE_TASK_STATUS_INTERRUPTED
             current_task.save()
         except FutureTask.DoesNotExist:
             pass
+
         self._running = False
 
     def _handle_options(self, options):
-        self.tick = 1
-        self.one_time_run = options["onetimerun"]
+        self.one_time_run = options["one_time_run"]
+        self.wait_for_tasks_duration_seconds = options["wait_for_tasks_duration_seconds"]
 
-    @staticmethod
-    def tasks_for_processing():
+    def _get_open_tasks(self):
         return FutureTask.objects.filter(
             eta__lte=timezone.now(),
             status=FutureTask.FUTURE_TASK_STATUS_OPEN,
         ).order_by("eta")
 
+    def _endless_task_iterator(self):
+        while self._running:
+            tasks = self._get_open_tasks()
+            yield from tasks
+            if not tasks:
+                time.sleep(self.wait_for_tasks_duration_seconds)
+
     @staticmethod
     def _convert_exception_args(args):
         return [str(arg) for arg in args]
 
-    def handle_tick(self):
-        task_list = self.tasks_for_processing()
-        logger.debug(f"Got {len(task_list)} tasks for processing")
+    def _handle_task(self, task):
+        task.status = FutureTask.FUTURE_TASK_STATUS_IN_PROGRESS
+        task.save()
+        self.current_task_pk = task.pk
+        try:
+            start_time = timeit.default_timer()
+            future_task_signal.send(sender=intern(task.type), instance=task)
+            task.execution_time = timeit.default_timer() - start_time
+            task.status = FutureTask.FUTURE_TASK_STATUS_DONE
+        except Exception as exception:
+            task.status = FutureTask.FUTURE_TASK_STATUS_ERROR
+            task.result = {
+                "exception": f"An exception of type {type(exception).__name__} occurred.",
+                "args": self._convert_exception_args(exception.args),
+                "traceback": traceback.format_exception(
+                    *sys.exc_info(),
+                    limit=None,
+                    chain=None,
+                ),
+            }
+            logger.exception(exception)
+        self.current_task_pk = None
+        task.save()
 
-        for task in task_list:
-            task.status = FutureTask.FUTURE_TASK_STATUS_IN_PROGRESS
-            task.save()
-            self.current_task_pk = task.pk
-            try:
-                start_time = timeit.default_timer()
-                future_task_signal.send(sender=intern(task.type), instance=task)
-                task.execution_time = timeit.default_timer() - start_time
-                task.status = FutureTask.FUTURE_TASK_STATUS_DONE
-            except Exception as exc:
-                task.status = FutureTask.FUTURE_TASK_STATUS_ERROR
-                task.result = {
-                    "exception": "An exception of type {} occurred.".format(
-                        type(exc).__name__,
-                    ),
-                    "args": self._convert_exception_args(exc.args),
-                    "traceback": traceback.format_exception(
-                        *sys.exc_info(),
-                        limit=None,
-                        chain=None,
-                    ),
-                }
-                logger.exception(exc)
-            self.current_task_pk = None
-            task.save()
-
-        time.sleep(self.tick)
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--one-time-run",
+            action="store_true",
+            help="Process tasks that are open at the time of running the command and exit.",
+        )
+        parser.add_argument(
+            "--wait-for-tasks-duration-seconds",
+            type=float,
+            default=1.0,
+            help="If there are no open tasks the command waits this amount of time until it checks for open tasks again.",
+        )
 
     def handle(self, *args, **options):
         # Load given options.
         self._handle_options(options)
-
+        tasks = iter(self._get_open_tasks()) if self.one_time_run else self._endless_task_iterator()
         while self._running:
-            time.sleep(self.tick)
-
             try:
-                self.handle_tick()
-                if self.one_time_run:
-                    break
-
+                self._handle_task(next(tasks))
+            except StopIteration:
+                break
             except Exception as exc:
                 logger.exception(
                     f"{exc.__class__.__name__} exception occurred...",
                 )
-
                 # As the database connection might have failed, we discard it here, so django will
                 # create a new one on the next database access.
                 db.close_old_connections()
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        # The command will run as long as the `_running` attribute is
-        # set to `True`. To safely quit the command, just set this attribute to `False` and the
-        # command will finish a running tick and quit afterwards.
-        self._running = True
-
-        # Register system signal handler to gracefully quit the service when
-        # getting a `SIGINT` or `SIGTERM` signal (e.g. by CTRL+C).
-        signal.signal(signal.SIGINT, self._handle_termination)
-        signal.signal(signal.SIGTERM, self._handle_termination)

django_future_tasks/migrations/0006_periodicfuturetask_end_time.py

@@ -1,5 +1,6 @@
 # Generated by Django 3.2.22 on 2023-10-20 15:42
 
+import django
 from django.db import migrations, models
 
 
@@ -27,6 +28,15 @@ class Migration(migrations.Migration):
                     _connector="OR",
                 ),
                 name="not_both_not_null",
+            )
+            if django.VERSION < (5, 1)
+            else models.CheckConstraint(
+                condition=models.Q(
+                    ("end_time__isnull", True),
+                    ("max_number_of_executions__isnull", True),
+                    _connector="OR",
+                ),
+                name="not_both_not_null",
             ),
         ),
     ]