207255_Celery_tasks_refactor

src/hope/admin/async_job.py

@@ -0,0 +1,193 @@
+from typing import Any, cast
+
+from adminfilters.autocomplete import AutoCompleteFilter
+from django.contrib import admin
+from django.db.models import QuerySet
+from django.http import HttpRequest
+
+from hope.admin.utils import HOPEModelAdminBase
+from hope.models import AsyncJob
+
+
+class HasErrorsListFilter(admin.SimpleListFilter):
+    title = "has errors"
+    parameter_name = "has_errors"
+
+    def lookups(
+        self, request: HttpRequest, model_admin: admin.ModelAdmin[Any]
+    ) -> tuple[tuple[str, str], tuple[str, str]]:
+        return (
+            ("yes", "Yes"),
+            ("no", "No"),
+        )
+
+    def queryset(self, request: HttpRequest, queryset: QuerySet[AsyncJob]) -> QuerySet[AsyncJob]:
+        if self.value() == "yes":
+            return queryset.exclude(errors={})
+        if self.value() == "no":
+            return queryset.filter(errors={})
+        return queryset
+
+
+@admin.register(AsyncJob)
+class AsyncJobAdmin(HOPEModelAdminBase):
+    list_display = (
+        "id",
+        "job_name",
+        "type",
+        "task_status_display",
+        "local_status",
+        "program",
+        "gfk_display",
+        "curr_async_result_id",
+        "repeatable",
+        "datetime_created",
+        "datetime_queued",
+    )
+    list_filter = (
+        ("program", AutoCompleteFilter),
+        ("content_type", AutoCompleteFilter),
+        ("job_name", AutoCompleteFilter),
+        HasErrorsListFilter,
+        "type",
+        "repeatable",
+        "local_status",
+    )
+    search_fields = (
+        "job_name",
+        "action",
+        "program__id",
+        "curr_async_result_id",
+        "last_async_result_id",
+        "object_id",
+    )
+    readonly_fields = (
+        "task_status_display",
+        "content_object_display",
+        "id",
+        "version",
+        "description",
+        "curr_async_result_id",
+        "last_async_result_id",
+        "datetime_created",
+        "datetime_queued",
+        "repeatable",
+        "celery_history",
+        "local_status",
+        "owner",
+        "group_key",
+        "type",
+        "config",
+        "action",
+        "sentry_id",
+        "errors",
+        "program",
+        "content_type",
+        "object_id",
+        "job_name",
+    )
+    ordering = ("-datetime_created",)
+
+    fieldsets = (
+        (
+            None,
+            {
+                "fields": (
+                    "id",
+                    "job_name",
+                    "type",
+                    "task_status_display",
+                    "local_status",
+                    "description",
+                    "action",
+                    "config",
+                    "errors",
+                )
+            },
+        ),
+        (
+            "Relations",
+            {
+                "fields": (
+                    "program",
+                    "owner",
+                    "content_type",
+                    "object_id",
+                    "content_object_display",
+                )
+            },
+        ),
+        (
+            "Execution",
+            {
+                "fields": (
+                    "repeatable",
+                    "group_key",
+                    "curr_async_result_id",
+                    "last_async_result_id",
+                    "datetime_created",
+                    "datetime_queued",
+                    "sentry_id",
+                    "celery_history",
+                    "version",
+                )
+            },
+        ),
+    )
+
+    @admin.display(description="Task status")
+    def task_status_display(self, obj: AsyncJob) -> str:
+        return obj.task_status
+
+    @admin.display(description="GFK")
+    def gfk_display(self, obj: AsyncJob) -> str:
+        if not obj.content_type_id or not obj.object_id:
+            return "-"
+        return f"{obj.content_type}:{obj.object_id}"
+
+    @admin.display(description="Content object")
+    def content_object_display(self, obj: AsyncJob) -> str:
+        return str(obj.content_object) if obj.content_object else "-"
+
+    def get_queryset(self, request: HttpRequest) -> QuerySet[AsyncJob]:
+        return cast(
+            "QuerySet[AsyncJob]",
+            super()
+            .get_queryset(request)
+            .select_related("program", "owner", "content_type")
+            .only(
+                "id",
+                "job_name",
+                "type",
+                "local_status",
+                "program",
+                "program__id",
+                "program__name",
+                "content_type",
+                "content_type__app_label",
+                "content_type__model",
+                "object_id",
+                "curr_async_result_id",
+                "repeatable",
+                "datetime_created",
+                "datetime_queued",
+                "description",
+                "action",
+                "config",
+                "errors",
+                "version",
+                "last_async_result_id",
+                "celery_history",
+                "owner",
+                "owner__id",
+                "owner__username",
+                "group_key",
+                "sentry_id",
+            ),
+        )
+
+    def has_add_permission(self, request: HttpRequest) -> bool:
+        return False
+
+    def has_delete_permission(self, request: HttpRequest, obj: Any | None = None) -> bool:
+        return False

src/hope/admin/geo.py

@@ -22,7 +22,7 @@
 from smart_admin.mixins import FieldsetMixin
 
 from hope.admin.utils import HOPEModelAdminBase
-from hope.apps.geo.celery_tasks import import_areas_from_csv_task
+from hope.apps.geo.celery_tasks import import_areas_from_csv_async_task
 from hope.models import Area, AreaType, Country
 
 if TYPE_CHECKING:
@@ -255,7 +255,7 @@ def import_areas(
                 existing_p_codes = set(Area.objects.filter(p_code__in=all_p_codes).values_list("p_code", flat=True))
                 new_areas_count = len(all_p_codes - existing_p_codes)
 
-                import_areas_from_csv_task.delay(data_set, delay_mptt_updates)
+                import_areas_from_csv_async_task(data_set, delay_mptt_updates)
 
                 self.message_user(
                     request,

src/hope/admin/household.py

@@ -15,7 +15,7 @@
 from django.contrib.messages import DEFAULT_TAGS
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import transaction
-from django.db.models import F, Q, QuerySet, Value
+from django.db.models import Case, F, Q, QuerySet, Value, When
 from django.db.transaction import atomic
 from django.forms import Form
 from django.http import HttpRequest, HttpResponse, HttpResponseBase, HttpResponseRedirect
@@ -37,8 +37,8 @@
 from hope.apps.grievance.signals import increment_grievance_ticket_version_cache_for_ticket_ids
 from hope.apps.household.api.caches import invalidate_household_and_individual_list_cache
 from hope.apps.household.celery_tasks import (
-    enroll_households_to_program_task,
-    mass_withdraw_households_from_list_task,
+    enroll_households_to_program_async_task,
+    mass_withdraw_households_from_list_async_task,
 )
 from hope.apps.household.forms import (
     MassEnrollForm,
@@ -244,31 +244,35 @@ def has_add_permission(self, request: HttpRequest, obj: Household | None = None)
 
 class HouseholdWithdrawFromListMixin:
     @staticmethod
-    def get_household_queryset_from_list(household_id_list: list, program: Program) -> QuerySet:
+    def get_household_queryset_from_list(household_id_list: list[str], program: Program) -> QuerySet:
         return Household.objects.filter(
             unicef_id__in=household_id_list,
             withdrawn=False,
             program=program,
         )
 
     @transaction.atomic
-    def mass_withdraw_households_from_list_bulk(self, household_id_list: list, tag: str, program: Program) -> None:
+    def mass_withdraw_households_from_list_bulk(self, household_id_list: list[str], tag: str, program: Program) -> None:
         households = self.get_household_queryset_from_list(household_id_list, program)
         individuals = Individual.objects.filter(household__in=households, withdrawn=False, duplicate=False)
 
         tickets = GrievanceTicket.objects.belong_households_individuals(households, individuals)
-        ticket_ids = [str(t.ticket.id) for t in tickets]
-        for status, _ in GrievanceTicket.STATUS_CHOICES:
-            if status == GrievanceTicket.STATUS_CLOSED:
-                continue
-            GrievanceTicket.objects.filter(id__in=ticket_ids, status=status).update(
-                extras=JSONBSet(
-                    F("extras"),
-                    Value("{status_before_withdrawn}"),
-                    Value(f'"{status}"'),
-                ),
-                status=GrievanceTicket.STATUS_CLOSED,
-            )
+        ticket_ids = list({t.ticket_id for t in tickets})
+        previous_status = Case(
+            *[
+                When(status=status, then=Value(f'"{status}"'))
+                for status, _ in GrievanceTicket.STATUS_CHOICES
+                if status != GrievanceTicket.STATUS_CLOSED
+            ]
+        )
+        GrievanceTicket.objects.filter(id__in=ticket_ids).exclude(status=GrievanceTicket.STATUS_CLOSED).update(
+            extras=JSONBSet(
+                F("extras"),
+                Value("{status_before_withdrawn}"),
+                previous_status,
+            ),
+            status=GrievanceTicket.STATUS_CLOSED,
+        )
         increment_grievance_ticket_version_cache_for_ticket_ids(program.business_area.slug, ticket_ids)
 
         Document.objects.filter(individual__in=individuals).update(status=Document.STATUS_INVALID)
@@ -350,7 +354,7 @@ def withdraw_households_from_list(self, request: HttpRequest) -> HttpResponse |
                 )
 
             if step == "3":
-                mass_withdraw_households_from_list_task.delay(household_id_list, tag, str(program.id))
+                mass_withdraw_households_from_list_async_task(household_id_list, tag, program)
                 self.message_user(request, f"{len(household_id_list)} Households are being withdrawn.")
                 return HttpResponseRedirect(reverse("admin:household_household_changelist"))
         return TemplateResponse(
@@ -639,9 +643,9 @@ def mass_enroll_to_another_program(self, request: HttpRequest, qs: QuerySet) ->
             if form.is_valid():
                 program_for_enroll = form.cleaned_data["program_for_enroll"]
                 households_ids = list(qs.distinct("unicef_id").values_list("id", flat=True))
-                enroll_households_to_program_task.delay(
+                enroll_households_to_program_async_task(
                     households_ids=households_ids,
-                    program_for_enroll_id=str(program_for_enroll.id),
+                    program_for_enroll_id=program_for_enroll,
                     user_id=str(request.user.id),
                 )
                 self.message_user(
@@ -663,7 +667,7 @@ def mass_enroll_to_another_program(self, request: HttpRequest, qs: QuerySet) ->
         context["action"] = "mass_enroll_to_another_program"
         return TemplateResponse(
             request,
-            "admin/household/household/enroll_households_to_program.html",
+            "admin/household/household/enroll_households_to_program_async_task.html",
             context,
         )
 

src/hope/admin/individual.py

@@ -27,7 +27,7 @@
     RdiMergeStatusAdminMixin,
     SoftDeletableAdminMixin,
 )
-from hope.apps.household.celery_tasks import revalidate_phone_number_task
+from hope.apps.household.celery_tasks import revalidate_phone_number_async_task
 from hope.apps.utils.security import is_root
 from hope.models import (
     Account,
@@ -227,17 +227,16 @@ def sanity_check(self, request: HttpRequest, pk: UUID) -> TemplateResponse:
 
     def revalidate_phone_number_sync(self, request: HttpRequest, queryset: QuerySet) -> None:
         try:
-            ids = queryset.values_list("id", flat=True)
-            revalidate_phone_number_task(ids)
+            ids = list(queryset.values_list("id", flat=True))
+            revalidate_phone_number_async_task(ids)
             self.message_user(request, f"Updated {len(ids)} records", messages.SUCCESS)
         except Error as e:
             self.message_user(request, str(e), messages.ERROR)
 
     revalidate_phone_number_sync.short_description = "Re-validate phone number (sync)"
 
     def revalidate_phone_number_async(self, request: HttpRequest, queryset: QuerySet) -> None:
-        ids = list(queryset.values_list("id", flat=True))
-        revalidate_phone_number_task.delay(ids)
+        revalidate_phone_number_async_task(list(queryset.values_list("id", flat=True)))
         self.message_user(request, "Updating in progress", messages.SUCCESS)
 
     revalidate_phone_number_async.short_description = "Re-validate phone number (async)"

src/hope/admin/kobo_template.py

@@ -14,14 +14,14 @@
     HttpResponsePermanentRedirect,
     HttpResponseRedirect,
 )
-from django.shortcuts import get_object_or_404, redirect
+from django.shortcuts import redirect
 from django.template.response import TemplateResponse
 from django.utils.html import format_html
 from openpyxl import load_workbook
 
 from hope.admin.utils import HOPEModelAdminBase, SoftDeletableAdminMixin
 from hope.apps.core.celery_tasks import (
-    upload_new_kobo_template_and_update_flex_fields_task,
+    upload_new_kobo_template_and_update_flex_fields_async_task,
 )
 from hope.apps.core.validators import KoboTemplateValidator
 from hope.models import XLSXKoboTemplate
@@ -101,10 +101,7 @@ def download_last_valid_file(
         permission="core.rerun_kobo_import",
     )
     def rerun_kobo_import(self, request: HttpRequest, pk: "UUID") -> HttpResponseBase | None:
-        xlsx_kobo_template_object = get_object_or_404(XLSXKoboTemplate, pk=pk)
-        upload_new_kobo_template_and_update_flex_fields_task.run(
-            xlsx_kobo_template_id=str(xlsx_kobo_template_object.id)
-        )
+        upload_new_kobo_template_and_update_flex_fields_async_task(xlsx_kobo_template_id=str(pk))
         return redirect(".")
 
     def add_view(
@@ -171,7 +168,7 @@ def add_view(
                     "Core field validation successful, running KoBo Template upload task..., "
                     "Import status will change after task completion",
                 )
-                upload_new_kobo_template_and_update_flex_fields_task.run(
+                upload_new_kobo_template_and_update_flex_fields_async_task(
                     xlsx_kobo_template_id=str(xlsx_kobo_template_object.id)
                 )
                 return redirect("..")

src/hope/admin/payment_plan.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any
 
 from admin_cursor_paginator import CursorPaginatorAdmin
 from admin_extra_buttons.decorators import button
@@ -222,7 +222,7 @@ def regenerate_export_xlsx(self, request: HttpRequest, pk: "UUID") -> HttpRespon
                 template_obj = form.cleaned_data.get("template")
                 fsp_xlsx_template_id = str(template_obj.id) if template_obj else None
                 PaymentPlanService(payment_plan=payment_plan).export_xlsx_per_fsp(
-                    cast("UUID", request.user.pk), fsp_xlsx_template_id
+                    str(request.user.pk), fsp_xlsx_template_id
                 )
                 messages.success(request, "Celery task for export regenerate file successfully started.")
                 return redirect(reverse("admin:payment_paymentplan_change", args=[pk]))