Enhance archive_feed command with force option

feeds/management/commands/archive_feed.py

@@ -2,6 +2,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 
 from django.core.management.base import BaseCommand
+from django.db.models.query import QuerySet
 
 from feeds.models import Entry
 from feeds.models import Feed
@@ -19,7 +20,7 @@ class Command(BaseCommand):
     amount_to_show: int = 10
 
     def add_arguments(self, parser: CommandParser) -> None:
-        """Add URL argument and --reset option to the command."""
+        """Add URL argument and options to the command."""
         parser.add_argument(
             "url",
             type=str,
@@ -30,11 +31,17 @@ class Command(BaseCommand):
             action="store_true",
             help="Remove all entries for this feed before archiving.",
         )
+        parser.add_argument(
+            "--force",
+            action="store_true",
+            help="Run the command non-interactively, skipping confirmations.",
+        )
 
     def handle(self, *args, **options) -> None:  # noqa: ARG002
         """Handle the command execution."""
         url: str = options["url"]
         reset: bool = options.get("reset", False)
+        force: bool = options.get("force", False)
 
         feed, created = Feed.objects.get_or_create(url=url)
 
@@ -51,32 +58,12 @@ class Command(BaseCommand):
                 self.stdout.write(self.style.WARNING(msg))
 
             else:
-                msg = f"The following {count} entries will be removed for feed: {url}"
+                if not force:
-                self.stdout.write(self.style.WARNING(msg))
+                    return self.confirm_and_list_entries(url, entries_qs, count)
 
-                entries = entries_qs.order_by("-published_at")[: self.amount_to_show]
+                entries_qs.delete()
-                for entry in entries:
+                msg = f"Deleted {count} entries for feed: {url}"
-                    title: str | None = get_entry_title(entry)
+                self.stdout.write(self.style.SUCCESS(msg))
-
-                    msg = f"- entry_id: {entry.entry_id}, published_at: {entry.published_at}, title: {title}"
-                    self.stdout.write(self.style.WARNING(msg))
-
-                if count > self.amount_to_show:
-                    self.stdout.write(f"...and {count - self.amount_to_show} more.")
-
-                prompt = "Are you sure you want to delete these entries? Type 'yes' to confirm: "
-                confirm: str = input(prompt)
-
-                if confirm.strip().lower() == "yes":
-                    deleted, _ = entries_qs.delete()
-
-                    msg = f"Deleted {deleted} entr{'y' if deleted == 1 else 'ies'} for feed: {url}"
-                    self.stdout.write(self.style.SUCCESS(msg))
-
-                else:
-                    msg = "Aborted reset. No entries were deleted."
-                    self.stdout.write(self.style.ERROR(msg))
-                    return
 
         new_entries: int = fetch_and_archive_feed(feed)
         if new_entries:
@@ -86,6 +73,28 @@ class Command(BaseCommand):
         else:
             msg: str = "\tFeed is up to date, but no new entries were archived."
             self.stdout.write(self.style.WARNING(msg))
+        return None
+
+    def confirm_and_list_entries(
+        self,
+        url: str,
+        entries_qs: QuerySet[Entry, Entry],
+        count: int,
+    ) -> None:
+        """Confirm with the user before deleting entries and list some of them."""
+        msg: str = f"The following {count} entries will be removed for feed: {url}"
+        self.stdout.write(self.style.WARNING(msg))
+
+        entries: QuerySet[Entry, Entry] = entries_qs.order_by("-published_at")
+        entries = entries[: self.amount_to_show]
+        for entry in entries:
+            title: str | None = get_entry_title(entry)
+            self.stdout.write(f"- {title}")
+
+        confirm: str = input("Are you sure you want to proceed? (yes/no): ")
+        if confirm.lower() != "yes":
+            self.stdout.write(self.style.ERROR("Operation cancelled."))
+            return
 
 
 def get_entry_title(entry: Entry) -> str | None:

feeds/tasks.py

@@ -1,14 +1,24 @@
+from typing import TYPE_CHECKING
+
 from celery import shared_task
 
 from feeds.models import Feed
 from feeds.services import fetch_and_archive_feed
 
+if TYPE_CHECKING:
+    from celery import Task
 
-@shared_task
+
-def archive_feed_task(feed_id: int) -> str:
+@shared_task(
+    bind=True,
+    autoretry_for=(Exception,),
+    retry_kwargs={"max_retries": 3, "countdown": 60},
+)
+def archive_feed_task(self: Task, feed_id: int) -> str:
     """Celery task to fetch and archive a feed by its ID.
 
     Args:
+        self: The task instance.
         feed_id: The ID of the Feed to archive.
 
     Returns:
@@ -18,7 +28,14 @@ def archive_feed_task(feed_id: int) -> str:
         feed: Feed = Feed.objects.get(id=feed_id)
     except Feed.DoesNotExist:
         return f"Feed with id {feed_id} does not exist."
-    new_entries_count: int = fetch_and_archive_feed(feed)
+
-    if new_entries_count > 0:
+    try:
-        return f"Archived {new_entries_count} new entries for {feed.url}"
+        new_entries_count: int = fetch_and_archive_feed(feed)
-    return f"No new entries archived for {feed.url}"
+
+    # TODO(TheLovinator): Replace with a specific exception type  # noqa: TD003
+    except ValueError as e:
+        raise self.retry(exc=e) from e
+    else:
+        if new_entries_count > 0:
+            return f"Archived {new_entries_count} new entries for {feed.url}"
+        return f"No new entries archived for {feed.url}"

feeds/urls.py

@@ -10,7 +10,8 @@ if TYPE_CHECKING:
 
 
 urlpatterns: list[URLPattern | URLResolver] = [
-    path("", views.feed_list, name="feed-list"),
+    path("", views.home, name="home"),
+    path("feeds/", views.feed_list, name="feed-list"),
     path("feeds/<int:feed_id>/", views.feed_detail, name="feed-detail"),
     path(
         "feeds/<int:feed_id>/entries/<int:entry_id>/",

feeds/views.py

@@ -1,9 +1,13 @@
+from __future__ import annotations
+
 import html
 import json
 from typing import TYPE_CHECKING
 
 from django.http import HttpResponse
 from django.shortcuts import get_object_or_404
+from django.shortcuts import redirect
+from django.shortcuts import render
 
 from feeds.models import Entry
 from feeds.models import Feed
@@ -20,17 +24,7 @@ def feed_list(request: HttpRequest) -> HttpResponse:
         HttpResponse: An HTML response containing the list of feeds.
     """
     feeds = Feed.objects.all().order_by("id")
-    html = [
+    return render(request, "feeds/feed_list.html", {"feeds": feeds})
-        "<!DOCTYPE html>",
-        "<html><head><title>FeedVault - Feeds</title></head><body>",
-        "<h1>Feed List</h1>",
-        "<ul>",
-    ]
-    html.extend(
-        f'<li><a href="/feeds/{feed.pk}/">{feed.url}</a></li>' for feed in feeds
-    )
-    html.extend(("</ul>", "</body></html>"))
-    return HttpResponse("\n".join(html))
 
 
 def feed_detail(request: HttpRequest, feed_id: int) -> HttpResponse:
@@ -115,3 +109,15 @@ def entry_detail(request: HttpRequest, feed_id: int, entry_id: int) -> HttpRespo
         "</body></html>",
     ]
     return HttpResponse("\n".join(html_lines))
+
+
+def home(request: HttpRequest) -> HttpResponse:
+    """Redirect to the feed list as the homepage.
+
+    Args:
+        request: The HTTP request object.
+
+    Returns:
+        HttpResponse: A redirect response to the feed list.
+    """
+    return redirect("/feeds/")

templates/base.html

@@ -0,0 +1,35 @@
+<!DOCTYPE html>
+<html lang="en">
+    <head>
+        <meta charset="UTF-8" />
+        <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+        <meta name="keywords" content="feeds, archive, FeedVault" />
+        <title>
+            {% block title %}
+                FeedVault
+            {% endblock title %}
+        </title>
+    </head>
+    <body>
+        <header>
+            <h1>FeedVault</h1>
+            <nav>
+                <ul>
+                    <li>
+                        <a href="{% url 'home' %}">Home</a>
+                    </li>
+                    <li>
+                        <a href="{% url 'feed-list' %}">Feeds</a>
+                    </li>
+                </ul>
+            </nav>
+        </header>
+        <main>
+            {% block content %}
+            {% endblock content %}
+        </main>
+        <footer>
+            <p>Web scraping is not a crime. - No rights reserved. - A birthday present for Plipp ❤️</p>
+        </footer>
+    </body>
+</html>

templates/feeds/feed_list.html

@@ -0,0 +1,11 @@
+{% extends "base.html" %}
+{% block content %}
+    <h1>Feed List</h1>
+    <ul>
+        {% for feed in feeds %}
+            <li>
+                <a href="{% url 'feed-detail' feed.pk %}">{{ feed.url }}</a>
+            </li>
+        {% endfor %}
+    </ul>
+{% endblock content %}