Add management commands to update unsplash photos every x days

website/contrib/unsplash/management/commands/update_unsplash_photos.py

@@ -0,0 +1,30 @@
+from argparse import ArgumentParser
+from datetime import timedelta
+
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+
+from website.contrib.unsplash.models import UnsplashPhoto
+from website.contrib.unsplash.utils import get_unsplash_photo
+from website.utils.queue import enqueue_or_sync
+
+
+def update_photo(photo: UnsplashPhoto) -> None:
+    photo.data = get_unsplash_photo(photo.unsplash_id)
+    photo.data_last_updated = timezone.now()
+    photo.save()
+
+
+class Command(BaseCommand):
+    def add_arguments(self, parser: ArgumentParser) -> None:
+        parser.add_argument("--max-age-days", type=int, default=10)
+
+    def handle(self, *args: list, **options: dict) -> None:
+        max_age_days: int = options["max_age_days"]  # type: ignore
+        max_age = timezone.now() - timedelta(days=max_age_days)
+        photos = UnsplashPhoto.objects.filter(data_last_updated__lte=max_age)
+        self.stdout.write(f"Found {photos.count()} photos to update.")
+
+        for photo in photos:
+            self.stdout.write(f"Updating {photo.unsplash_id}")
+            enqueue_or_sync(update_photo, args=[photo])

website/utils/queue.py

@@ -0,0 +1,21 @@
+from typing import Callable
+
+from django.conf import settings
+from django_rq import get_queue
+
+
+def enqueue_or_sync(
+    job_func: Callable, args: list | None = None, kwargs: dict | None = None
+) -> None:
+    """
+    Run a task now, or put in RQ
+    """
+    if args is None:
+        args = []
+    if kwargs is None:
+        kwargs = {}
+
+    if settings.USE_REDIS_QUEUE:
+        get_queue().enqueue(job_func, args=args, kwargs=kwargs)
+    else:
+        job_func(*args, **kwargs)