feat(emerg): move csl status checker to celerybeat task

intranet/apps/emerg/tasks.py

@@ -1,5 +1,10 @@
+import requests
 from celery import shared_task
 from celery.utils.log import get_task_logger
+from django.conf import settings
+from django.core.cache import cache
+from requests.adapters import HTTPAdapter
+from urllib3 import Retry
 
 from .views import update_emerg_cache
 
@@ -10,3 +15,30 @@
 def update_emerg_cache_task() -> None:
     logger.debug("Updating FCPS emergency info")
     update_emerg_cache(custom_logger=logger)
+
+
+@shared_task
+def update_csl_status_task() -> None:
+    """Updates the cached status of the tjCSL status page.
+
+    Returns:
+        Nothing
+    """
+    logger.debug("Updating CSL Status")
+    session = requests.Session()
+    adapter = HTTPAdapter(
+        max_retries=Retry(
+            total=settings.CSL_STATUS_PAGE_MAX_RETRIES, backoff_factor=0.3, status_forcelist=[500, 502, 503, 504], allowed_methods=["GET"]
+        )
+    )
+    session.mount("https://", adapter)
+
+    try:
+        response = session.get(settings.CSL_STATUS_PAGE, timeout=settings.CSL_STATUS_PAGE_TIMEOUT)
+        response.raise_for_status()
+        status = response.json()["data"]["attributes"]["aggregate_state"]
+    except Exception as ex:
+        status = "error"
+        logger.error(f"Could not fetch status page or incorrect status page JSON format: {ex}")
+
+    cache.set("emerg:csl_status", status, settings.CACHE_AGE["csl_status"])

intranet/apps/emerg/views.py

@@ -1,13 +1,12 @@
 import logging
 import time
+from typing import Literal
 
 import requests
 from bs4 import BeautifulSoup
 from django.conf import settings
 from django.core.cache import cache
 from django.utils import timezone
-from requests.adapters import HTTPAdapter
-from urllib3 import Retry
 
 from ...utils.html import get_domain_name, safe_fcps_emerg_html
 
@@ -123,38 +122,14 @@ def update_emerg_cache(*, custom_logger=None) -> None:
     cache.set(key, result, timeout=settings.CACHE_AGE["emerg"])
 
 
-def get_csl_status() -> tuple[str, bool]:
+def get_csl_status() -> Literal["error", "operational", "downtime", "degraded", "maintenance"]:
     """Get the cached status of the TJCSL status page.
 
     Returns:
-        Tuple with a string consisting of the aggregate status
-        of the TJ computer systems lab and a bool indicating whether
-        the status cache was updated
+        A string consisting of the aggregate status
+        of the TJ computer systems lab.
 
         The string of the tuple will be one of the following: "error" (parse error), "operational", "downtime", "degraded", "maintenance"
     """
 
-    status = cache.get("emerg:csl_status")
-    updated = False
-
-    if not status:
-        session = requests.Session()
-        adapter = HTTPAdapter(
-            max_retries=Retry(
-                total=settings.CSL_STATUS_PAGE_MAX_RETRIES, backoff_factor=0.3, status_forcelist=[500, 502, 503, 504], allowed_methods=["GET"]
-            )
-        )
-        session.mount("https://", adapter)
-
-        try:
-            response = session.get(settings.CSL_STATUS_PAGE, timeout=settings.CSL_STATUS_PAGE_TIMEOUT)
-            response.raise_for_status()
-            status = response.json()["data"]["attributes"]["aggregate_state"]
-            updated = True
-        except Exception as ex:
-            status = "error"
-            logger.error(f"Could not fetch status page or incorrect status page JSON format: {ex}")
-
-        cache.set("emerg:csl_status", status, settings.CACHE_AGE["csl_status"])
-
-    return status, updated
+    return cache.get("emerg:csl_status", "error")

intranet/apps/templatetags/status_helper.py

@@ -16,7 +16,7 @@ def get_cache(key):
 
 class GetCSLStatusNode(template.Node):
     def render(self, context):
-        context["csl_status"] = get_csl_status()[0]
+        context["csl_status"] = get_csl_status()
         return ""
 
 

intranet/settings/__init__.py

@@ -535,7 +535,7 @@ def get_month_seconds():
     "users_list": int(datetime.timedelta(hours=24).total_seconds()),
     "printers_list": int(datetime.timedelta(minutes=10).total_seconds()),
     "emerg": int(datetime.timedelta(minutes=5).total_seconds()),
-    "csl_status": int(datetime.timedelta(minutes=5).total_seconds()),
+    "csl_status": int(datetime.timedelta(minutes=10).total_seconds()),
     "sports_school_events": int(datetime.timedelta(hours=1).total_seconds()),
 }
 
@@ -963,6 +963,11 @@ def get_log(name):  # pylint: disable=redefined-outer-name; 'name' is used as th
         "schedule": FCPS_EMERGENCY_CACHE_UPDATE_INTERVAL,
         "args": (),
     },
+    "update-csl-status-cache": {
+        "task": "intranet.apps.emerg.tasks.update_csl_status_task",
+        "schedule": celery.schedules.crontab(minute="*/5"),
+        "args": (),
+    },
     "reset-routes-afternoon": {
         "task": "intranet.apps.bus.tasks.reset_routes",
         "schedule": celery.schedules.crontab(hour=0, minute=0),