Resolve CodeQL warnings & drop Python 3.8

Author: codingjoe

.github/workflows/ci.yml

@@ -59,7 +59,6 @@ jobs:
     strategy:
       matrix:
         python-version:
-          - "3.8"
           - "3.9"
           - "3.10"
           - "3.11"

.readthedocs.yaml

@@ -7,7 +7,7 @@ version: 2
 build:
   os: ubuntu-20.04
   tools:
-    python: "3.9"
+    python: "3.11"
   apt_packages:
     - graphviz
 

docs/conf.py

@@ -50,14 +50,10 @@ def linkcode_resolve(domain, info):
         try:
             lines, first_line = inspect.getsourcelines(item)
             lineno = "#L%d-L%s" % (first_line, first_line + len(lines) - 1)
-        except (TypeError, IOError):
+        except (TypeError, OSError):
             pass
-    return "https://github.com/%s/%s/blob/%s/%s.py%s" % (
-        github_user,
-        project,
-        head,
-        filename,
-        lineno,
+    return (
+        f"https://github.com/{github_user}/{project}/blob/{head}/{filename}.py{lineno}"
     )
 
 

joeflow/__init__.py

@@ -1,10 +1,5 @@
 """The lean workflow automation framework for machines with heart."""
-import django
-
 from . import _version
 
 __version__ = _version.version
 VERSION = _version.version_tuple
-
-if django.VERSION < (3, 2):
-    default_app_config = "joeflow.apps.JoeflowConfig"

joeflow/admin.py

@@ -10,6 +10,10 @@
 __all__ = ("WorkflowAdmin",)
 
 
+@admin.action(
+    description=t("Rerun selected tasks"),
+    permissions=("rerun",),
+)
 def rerun(modeladmin, request, queryset):
     succeeded = queryset.succeeded().count()
     if succeeded:
@@ -24,10 +28,10 @@ def rerun(modeladmin, request, queryset):
     messages.success(request, "%s tasks have been successfully queued" % counter)
 
 
-rerun.short_description = t("Rerun selected tasks")
-rerun.allowed_permissions = ("rerun",)
-
-
+@admin.action(
+    description=t("Cancel selected tasks"),
+    permissions=("cancel",),
+)
 def cancel(modeladmin, request, queryset):
     not_scheduled = queryset.not_scheduled().count()
     if not_scheduled:
@@ -40,32 +44,26 @@ def cancel(modeladmin, request, queryset):
     messages.success(request, "Tasks have been successfully canceled")
 
 
-cancel.short_description = t("Cancel selected tasks")
-cancel.allowed_permissions = ("cancel",)
-
-
 @admin.register(models.Task)
 class TaskAdmin(VersionAdmin):
     def has_rerun_permission(self, request):
         opts = self.opts
         codename = get_permission_codename("rerun", opts)
-        return request.user.has_perm("%s.%s" % (opts.app_label, codename))
+        return request.user.has_perm(f"{opts.app_label}.{codename}")
 
     def has_cancel_permission(self, request):
         opts = self.opts
         codename = get_permission_codename("cancel", opts)
-        return request.user.has_perm("%s.%s" % (opts.app_label, codename))
+        return request.user.has_perm(f"{opts.app_label}.{codename}")
 
+    @admin.display(description=t("Traceback"))
     def pretty_stacktrace(self, obj):
         return format_html('<pre class="readonly collapse">{}<pre>', obj.stacktrace)
 
-    pretty_stacktrace.short_description = t("Traceback")
-
+    @admin.display(description=t("Child tasks"))
     def child_tasks(self, obj):
         return ", ".join(str(task) for task in obj.child_task_set.all().iterator())
 
-    child_tasks.short_description = t("Child tasks")
-
     actions = (rerun, cancel)
 
     list_display = (

joeflow/contrib/reversion.py

@@ -1,6 +1,6 @@
 from contextlib import contextmanager
 
-from joeflow import utils
+import joeflow.models
 
 __all__ = (
     "register_workflows",
@@ -16,7 +16,7 @@ def register_workflows():
     except ImportError:
         pass
     else:
-        for workflow in utils.get_workflows():
+        for workflow in joeflow.models.get_workflows():
             if not revisions.is_registered(workflow):
                 revisions.register(workflow)
 

joeflow/forms.py

@@ -2,6 +2,7 @@
 from django.utils.translation import gettext_lazy as t
 
 from . import models
+from .typing import HUMAN
 
 
 class OverrideForm(forms.ModelForm):
@@ -35,7 +36,7 @@ def start_next_tasks(self, user=None):
                 parent_tasks = []
         override_task = self.instance.task_set.create(
             name="override",
-            type=models.Task.HUMAN,
+            type=HUMAN,
         )
         override_task.parent_task_set.set(parent_tasks)
         override_task.finish(user=user)

joeflow/management/commands/render_workflow_graph.py

@@ -1,7 +1,6 @@
 from django.core.management import BaseCommand
 
-from joeflow import utils
-from joeflow.models import Workflow
+import joeflow.models
 
 
 class Command(BaseCommand):
@@ -45,10 +44,12 @@ def handle(self, *args, **options):
         cleanup = options["cleanup"]
         directory = options.get("directory", None)
 
-        workflows = [utils.get_workflow(s) for s in workflows] or utils.get_workflows()
+        workflows = [
+            joeflow.models.get_workflow(s) for s in workflows
+        ] or joeflow.models.get_workflows()
 
         for workflow in filter(None, workflows):
-            if workflow != Workflow:
+            if workflow != joeflow.models.Workflow:
                 opt = workflow._meta
                 if verbosity > 0:
                     self.stdout.write(

joeflow/models.py

@@ -1,7 +1,8 @@
 import logging
 import sys
 import traceback
-from typing import Any, List, Tuple
+import types
+import typing
 
 from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
 from django.db import models, transaction
@@ -15,8 +16,8 @@
 from django.views import View
 from django.views.generic.edit import BaseCreateView
 
-from . import utils
 from .conf import settings
+from .typing import HUMAN, MACHINE
 from .utils import NoDashDiGraph
 
 logger = logging.getLogger(__name__)
@@ -42,10 +43,10 @@ def __new__(mcs, name, bases, attrs):
                 if func in nodes:
                     node = getattr(klass, name)
                     node.name = name
-                    node.type = getattr(node, "type", "machine")
+                    node.type = getattr(node, "type", MACHINE)
                     node.workflow_cls = klass
             except TypeError:
-                pass
+                pass  # not a function
         if "override_view" in attrs and isinstance(klass.override_view, str):
             klass.override_view = import_string(klass.override_view)
         if "detail_view" in attrs and isinstance(klass.detail_view, str):
@@ -82,7 +83,7 @@ def save(self, **kwargs):
                 update_fields.append("modified")
         super().save(**kwargs)
 
-    edges: List[Tuple[Any, Any]] = None
+    edges: list[tuple[typing.Any, typing.Any]] = None
     """
     Edges define the transitions between tasks.
 
@@ -136,9 +137,9 @@ def urls(cls):
         for name, node in cls.get_nodes():
             if isinstance(node, View):
                 if isinstance(node, BaseCreateView):
-                    route = "{name}/".format(name=name)
+                    route = f"{name}/"
                 else:
-                    route = "{name}/<int:pk>/".format(name=name)
+                    route = f"{name}/<int:pk>/"
                 urls.append(
                     path(
                         route + node.path,
@@ -177,15 +178,11 @@ def get_url_namespace(cls):
 
     def get_absolute_url(self):
         """Return URL to workflow detail view."""
-        return reverse(
-            "{}:detail".format(self.get_url_namespace()), kwargs=dict(pk=self.pk)
-        )
+        return reverse(f"{self.get_url_namespace()}:detail", kwargs=dict(pk=self.pk))
 
     def get_override_url(self):
         """Return URL to workflow override view."""
-        return reverse(
-            "{}:override".format(self.get_url_namespace()), kwargs=dict(pk=self.pk)
-        )
+        return reverse(f"{self.get_url_namespace()}:override", kwargs=dict(pk=self.pk))
 
     @classmethod
     def get_graph(cls, color="black"):
@@ -206,7 +203,7 @@ def get_graph(cls, color="black"):
         )
         for name, node in cls.get_nodes():
             node_style = "filled"
-            if node.type == "human":
+            if node.type == HUMAN:
                 node_style += ", rounded"
             graph.node(name, style=node_style, color=color, fontcolor=color)
 
@@ -252,7 +249,7 @@ def get_instance_graph(self):
             style = "filled"
             peripheries = "1"
 
-            if task.type == "human":
+            if task.type == HUMAN:
                 style += ", rounded"
             if not task.completed:
                 style += ", bold"
@@ -285,7 +282,7 @@ def get_instance_graph(self):
         for task in self.task_set.exclude(name__in=names).exclude(name="override"):
             style = "filled, dashed"
             peripheries = "1"
-            if task.type == "human":
+            if task.type == HUMAN:
                 style += ", rounded"
             if not task.completed:
                 style += ", bold"
@@ -340,7 +337,7 @@ def workflow_state_subclasses():
 
     apps.check_models_ready()
     query = models.Q()
-    for workflow in utils.get_workflows():
+    for workflow in get_workflows():
         opts = workflow._meta
         query |= models.Q(app_label=opts.app_label, model=opts.model_name)
     return query
@@ -396,8 +393,6 @@ class Task(models.Model):
 
     name = models.CharField(max_length=255, db_index=True, editable=False)
 
-    HUMAN = "human"
-    MACHINE = "machine"
     _type_choices = (
         (HUMAN, t(HUMAN)),
         (MACHINE, t(MACHINE)),
@@ -470,7 +465,7 @@ class Meta:
         default_manager_name = "objects"
 
     def __str__(self):
-        return "%s (%s)" % (self.name, self.pk)
+        return f"{self.name} ({self.pk})"
 
     def save(self, **kwargs):
         if self.pk:
@@ -486,12 +481,12 @@ def save(self, **kwargs):
 
     def get_absolute_url(self):
         if self.completed:
-            return
-        url_name = "{}:{}".format(self.workflow.get_url_namespace(), self.name)
+            return  # completed tasks have no detail view
+        url_name = f"{self.workflow.get_url_namespace()}:{self.name}"
         try:
             return reverse(url_name, kwargs=dict(pk=self.pk))
         except NoReverseMatch:
-            pass
+            return  # no URL was defined for this task
 
     @property
     def node(self):
@@ -582,3 +577,23 @@ def start_next_tasks(self, next_nodes: list = None):
                 transaction.on_commit(task.enqueue)
             tasks.append(task)
         return tasks
+
+
+def get_workflows() -> types.GeneratorType:
+    """Return all registered workflows."""
+    from django.apps import apps
+
+    apps.check_models_ready()
+    for model in apps.get_models():
+        if issubclass(model, Workflow) and model is not Workflow and model.edges:
+            yield model
+    return  # empty generator
+
+
+def get_workflow(name) -> typing.Optional[Workflow]:
+    for workflow_cls in get_workflows():
+        if (
+            name.lower()
+            == f"{workflow_cls._meta.app_label}.{workflow_cls.__name__}".lower()
+        ):
+            return workflow_cls

joeflow/tasks/__init__.py

@@ -4,8 +4,7 @@
 A task can be considered as a simple transaction that changes state of a workflow.
 There are two types of tasks, human and machine tasks.
 """
+from joeflow.typing import *  # NoQA
+
 from .human import *  # NoQA
 from .machine import *  # NoQA
-
-HUMAN = "human"
-MACHINE = "machine"

joeflow/tasks/machine.py

@@ -1,5 +1,5 @@
 """Set of reusable machine tasks."""
-from typing import Iterable
+from collections.abc import Iterable
 
 from django.utils import timezone
 

joeflow/typing.py

@@ -0,0 +1,2 @@
+HUMAN = "human"
+MACHINE = "machine"

joeflow/utils.py

@@ -1,30 +1,8 @@
-import types
 from collections import defaultdict
 
 import graphviz as gv
 
 
-def get_workflows() -> types.GeneratorType:
-    """Return all registered workflows."""
-    from django.apps import apps
-
-    from .models import Workflow
-
-    apps.check_models_ready()
-    for model in apps.get_models():
-        if issubclass(model, Workflow) and model is not Workflow and model.edges:
-            yield model
-
-
-def get_workflow(name):
-    for workflow_cls in get_workflows():
-        if (
-            name.lower()
-            == f"{workflow_cls._meta.app_label}.{workflow_cls.__name__}".lower()
-        ):
-            return workflow_cls
-
-
 class NoDashDiGraph(gv.Digraph):
     """
     Like `.graphviz.Digraph` but with unique nodes and edges.