Add mypy and pyright checking

.github/workflows/run-tests.yml

@@ -59,6 +59,8 @@ jobs:
         black --check . --exclude "/node_modules/"
         flake8 --extend-ignore=E501 --exclude=node_modules .
         isort --check-only --diff --profile=black --skip node_modules .
+        pyright .
+        mypy .
         npm run lint
         coverage run --include="./*" --omit="*/tests/*","*/tests.py","*/migrations/*","./enhydris_project/*","*.pyx" manage.py test -v2
         coverage json

doc/conf.py

@@ -1,6 +1,4 @@
-# -*- coding: utf-8 -*-
-#
-extensions = []
+extensions: list[str] = []
 templates_path = ["_templates"]
 source_suffix = ".rst"
 master_doc = "index"

enhydris/admin/__init__.py

@@ -1,16 +1,33 @@
+from typing import TYPE_CHECKING
+
 from django.contrib import admin
 from django.utils.translation import gettext_lazy as _
 
 from enhydris import models
 
-from .garea import GareaAdmin  # NOQA
-from .station import StationAdmin  # NOQA
+from .garea import GareaAdmin  # NOQA  # type: ignore
+from .station import StationAdmin  # NOQA  # type: ignore
 
 admin.site.site_header = _("Enhydris dashboard")
 
+if TYPE_CHECKING:
+    LentityAdminBase = admin.ModelAdmin[models.Lentity]
+    EventTypeAdminBase = admin.ModelAdmin[models.EventType]
+    VariableTranslationInlineBase = admin.TabularInline[
+        models.VariableTranslation, models.Variable
+    ]
+    VariableAdminBase = admin.ModelAdmin[models.Variable]
+    UnitOfMeasurementAdminBase = admin.ModelAdmin[models.UnitOfMeasurement]
+else:
+    LentityAdminBase = admin.ModelAdmin
+    EventTypeAdminBase = admin.ModelAdmin
+    VariableTranslationInlineBase = admin.TabularInline
+    VariableAdminBase = admin.ModelAdmin
+    UnitOfMeasurementAdminBase = admin.ModelAdmin
+
 
 @admin.register(models.Lentity)
-class LentityAdmin(admin.ModelAdmin):
+class LentityAdmin(LentityAdminBase):
     pass
 
 
@@ -25,21 +42,21 @@ class PersonAdmin(LentityAdmin):
 
 
 @admin.register(models.EventType)
-class EventTypeAdmin(admin.ModelAdmin):
+class EventTypeAdmin(EventTypeAdminBase):
     list_display = ("id", "descr")
 
 
-class VariableTranslationInline(admin.TabularInline):
+class VariableTranslationInline(VariableTranslationInlineBase):
     model = models.VariableTranslation
     extra = 1
 
 
 @admin.register(models.Variable)
-class VariableAdmin(admin.ModelAdmin):
+class VariableAdmin(VariableAdminBase):
     list_display = ("id", "descr", "last_modified")
-    inlines = [VariableTranslationInline]
+    inlines = (VariableTranslationInline,)
 
 
 @admin.register(models.UnitOfMeasurement)
-class UnitOfMeasurementAdmin(admin.ModelAdmin):
+class UnitOfMeasurementAdmin(UnitOfMeasurementAdminBase):
     list_display = [f.name for f in models.UnitOfMeasurement._meta.fields]

enhydris/admin/garea.py

@@ -1,29 +1,43 @@
+from __future__ import annotations
+
 import os
 import tempfile
 import textwrap
+from typing import TYPE_CHECKING
 from zipfile import BadZipFile, ZipFile
 
 from django import forms
 from django.contrib import messages
 from django.contrib.gis import admin
 from django.contrib.gis.forms import MultiPolygonField, OSMWidget
 from django.contrib.gis.gdal import DataSource
+from django.contrib.gis.gdal.feature import Feature
 from django.contrib.gis.geos import MultiPolygon
+from django.core.files.uploadedfile import UploadedFile
 from django.db import IntegrityError, transaction
-from django.http import HttpResponseRedirect
+from django.http import HttpRequest, HttpResponseRedirect
 from django.template.response import TemplateResponse
 from django.urls import path
 from django.utils.translation import gettext_lazy as _
 
 from enhydris import models
 
+if TYPE_CHECKING:
+    GareaCategoryAdminBase = admin.ModelAdmin[models.GareaCategory]
+    GareaFormBase = forms.ModelForm[models.Garea]
+    GareaAdminBase = admin.ModelAdmin[models.Garea]
+else:
+    GareaCategoryAdminBase = admin.ModelAdmin
+    GareaFormBase = forms.ModelForm
+    GareaAdminBase = admin.ModelAdmin
+
 
 class MissingAttribute(Exception):
     pass
 
 
 @admin.register(models.GareaCategory)
-class GareaCategory(admin.ModelAdmin):
+class GareaCategory(GareaCategoryAdminBase):
     pass
 
 
@@ -65,7 +79,7 @@ def clean_file(self):
         return data
 
 
-class GareaForm(forms.ModelForm):
+class GareaForm(GareaFormBase):
     geometry = MultiPolygonField(
         widget=OSMWidget,
         disabled=True,
@@ -84,7 +98,7 @@ class Meta:
 
 
 @admin.register(models.Garea)
-class GareaAdmin(admin.ModelAdmin):
+class GareaAdmin(GareaAdminBase):
     form = GareaForm
     list_display = ["id", "name", "code", "category"]
     list_filter = ["category"]
@@ -95,25 +109,25 @@ def get_urls(self):
         new_urls = [path("bulk_add/", self.admin_site.admin_view(self.bulk_add))]
         return new_urls + urls
 
-    def bulk_add(self, request):
+    def bulk_add(self, request: HttpRequest):
         if request.method == "POST":
             return self._bulk_post(request)
         else:
             return self._get_template_response(request, GareaUploadForm())
 
-    def _bulk_post(self, request):
+    def _bulk_post(self, request: HttpRequest):
         form = GareaUploadForm(request.POST, request.FILES)
         if form.is_valid():
             return self._process_uploaded_form(request, form)
         else:
             return self._get_template_response(request, form)
 
-    def _process_uploaded_form(self, request, form):
+    def _process_uploaded_form(self, request: HttpRequest, form: GareaUploadForm):
         try:
             category = models.GareaCategory.objects.get(id=request.POST["category"])
-            nnew, nold = self._process_uploaded_shapefile(
-                category, request.FILES["file"]
-            )
+            uploaded_file = request.FILES["file"]
+            assert isinstance(uploaded_file, UploadedFile)
+            nnew, nold = self._process_uploaded_shapefile(category, uploaded_file)
         except IntegrityError as e:
             messages.add_message(request, messages.ERROR, str(e))
         else:
@@ -126,13 +140,15 @@ def _process_uploaded_form(self, request, form):
             )
         return HttpResponseRedirect("")
 
-    def _get_template_response(self, request, form):
+    def _get_template_response(self, request: HttpRequest, form: GareaUploadForm):
         return TemplateResponse(
             request, "admin/enhydris/garea/bulk_add.html", {"form": form}
         )
 
     @transaction.atomic
-    def _process_uploaded_shapefile(self, category, file):
+    def _process_uploaded_shapefile(
+        self, category: models.GareaCategory, file: UploadedFile
+    ):
         zipfile = ZipFile(file)
         shapefilename = [x for x in zipfile.namelist() if x.lower()[-4:] == ".shp"][0]
         with tempfile.TemporaryDirectory() as tmpdir:
@@ -152,18 +168,20 @@ def _process_uploaded_shapefile(self, category, file):
                 nnew += 1
         return nnew, nold
 
-    def _get_garea(self, feature, category):
+    def _get_garea(self, feature: Feature, category: models.GareaCategory):
         garea = models.Garea()
         if isinstance(feature.geom.geos, MultiPolygon):
             garea.geom = feature.geom.geos
         else:
             garea.geom = MultiPolygon(feature.geom.geos)
-        garea.name = self._get_feature_attr(feature, "Name")
+        name = self._get_feature_attr(feature, "Name")
+        assert name is not None
+        garea.name = name
         garea.code = self._get_feature_attr(feature, "Code", allow_empty=True) or ""
         garea.category = category
         return garea
 
-    def _get_feature_attr(self, feature, attr, allow_empty=False):
+    def _get_feature_attr(self, feature: Feature, attr: str, allow_empty: bool = False):
         try:
             value = feature.get(attr)
         except IndexError:

enhydris/admin/station.py

@@ -10,10 +10,9 @@
 from django.db.models import Q, TextField
 from django.utils.translation import gettext_lazy as _
 
-import nested_admin
+import nested_admin  # type: ignore
 import pandas as pd
-from crequest.middleware import CrequestMiddleware
-from geowidgets import LatLonField
+from geowidgets import LatLonField  # type: ignore
 from htimeseries import HTimeseries
 from rules.contrib.admin import ObjectPermissionsModelAdmin
 
@@ -102,7 +101,7 @@ class GentityEventInline(InlinePermissionsMixin, nested_admin.NestedStackedInlin
 
 
 class TimeseriesInlineAdminForm(forms.ModelForm):
-    data = forms.FileField(label=_("Data file"), required=False)
+    data_file = forms.FileField(label=_("Data file"), required=False)
     default_timezone = forms.ChoiceField(
         label=_("Time zone"),
         required=False,
@@ -131,10 +130,14 @@ class Meta:
         model = models.Timeseries
         exclude = ()
 
+    def __init__(self, *args, request=None, **kwargs):
+        self.request = request
+        super().__init__(*args, **kwargs)
+
     def clean(self):
         result = super().clean()
-        if self.cleaned_data.get("data") is not None:
-            self._check_submitted_data(self.cleaned_data["data"])
+        if self.cleaned_data.get("data_file") is not None:
+            self._check_submitted_data(self.cleaned_data["data_file"])
         return result
 
     def clean_time_step(self):
@@ -195,17 +198,18 @@ def _check_timeseries_for_appending(self, ahtimeseries):
 
     def save(self, *args, **kwargs):
         result = super().save(*args, **kwargs)
-        if self.cleaned_data.get("data") is not None:
+        if self.cleaned_data.get("data_file") is not None:
             self._save_timeseries_data()
         return result
 
     def _save_timeseries_data(self):
-        request = CrequestMiddleware.get_request()
+        request = self.request
+        assert request is not None
 
         # Create a hard link to the temporary uploaded file with the data
         # so that it's not automatically deleted and can be used by the celery
         # worker
-        tmpfilename = self.cleaned_data["data"].temporary_file_path()
+        tmpfilename = self.cleaned_data["data_file"].temporary_file_path()
         datafilename = tmpfilename + ".1"
         os.link(tmpfilename, datafilename)
 
@@ -231,6 +235,14 @@ def _save_timeseries_data(self):
 
 
 class TimeseriesInlineFormSet(nested_admin.formsets.NestedInlineFormSet):
+    def __init__(self, *args, request=None, **kwargs):
+        self.request = request
+        super().__init__(*args, **kwargs)
+
+    def _construct_form(self, i, **kwargs):
+        kwargs["request"] = self.request
+        return super()._construct_form(i, **kwargs)
+
     def clean(self):
         super().clean()
         self._check_only_one_timeseries_with_type(models.Timeseries.INITIAL)
@@ -262,9 +274,19 @@ class TimeseriesInline(InlinePermissionsMixin, nested_admin.NestedStackedInline)
     fields = (
         ("type", "time_step", "name"),
         "publicly_available",
-        ("data", "default_timezone", "replace_or_append"),
+        ("data_file", "default_timezone", "replace_or_append"),
     )
 
+    def get_formset(self, request, obj=None, **kwargs):
+        base_formset = super().get_formset(request, obj, **kwargs)
+
+        class RequestAwareFormSet(base_formset):
+            def __init__(self, *args, **inner_kwargs):
+                inner_kwargs["request"] = request
+                super().__init__(*args, **inner_kwargs)
+
+        return RequestAwareFormSet
+
 
 class TimeseriesGroupInline(InlinePermissionsMixin, nested_admin.NestedStackedInline):
     model = models.TimeseriesGroup