Create task to purge old announcements

database/crud/ufora_announcements.py

@@ -1,6 +1,6 @@
-from datetime import datetime
+import datetime
 
-from sqlalchemy import select
+from sqlalchemy import select, delete
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import UforaCourse, UforaAnnouncement
@@ -8,8 +8,8 @@ from database.models import UforaCourse, UforaAnnouncement
 
 async def get_courses_with_announcements(session: AsyncSession) -> list[UforaCourse]:
     """Get all courses where announcements are enabled"""
-    query = select(UforaCourse).where(UforaCourse.log_announcements)
-    return (await session.execute(query)).scalars().all()
+    statement = select(UforaCourse).where(UforaCourse.log_announcements)
+    return (await session.execute(statement)).scalars().all()
 
 
 async def create_new_announcement(
@@ -22,3 +22,14 @@ async def create_new_announcement(
     session.add(new_announcement)
     await session.commit()
     return new_announcement
+
+
+async def remove_old_announcements(session: AsyncSession):
+    """Delete all announcements that are > 8 days old
+    The RSS feed only goes back 7 days, so all of these old announcements never have to
+    be checked again when checking if an announcement is fresh or not.
+    """
+    limit = datetime.datetime.utcnow() - datetime.timedelta(days=8)
+    statement = delete(UforaAnnouncement).where(UforaAnnouncement.publication_date < limit)
+    await session.execute(statement)
+    await session.commit()

didier/cogs/tasks.py

@@ -3,22 +3,23 @@ import traceback
 from discord.ext import commands, tasks
 
 import settings
+from database.crud.ufora_announcements import remove_old_announcements
 from didier import Didier
 from didier.data.embeds.ufora.announcements import fetch_ufora_announcements
 
 
-# TODO task to clean up old announcements? (> 1 week)
 class Tasks(commands.Cog):
     """Task loops that run periodically"""
 
     client: Didier
 
-    def __init__(self, client: Didier):
+    def __init__(self, client: Didier):  # pylint: disable=no-member
         self.client = client
 
         # Only pull announcements if a token was provided
         if settings.UFORA_RSS_TOKEN is not None and settings.UFORA_ANNOUNCEMENTS_CHANNEL is not None:
-            self.pull_ufora_announcements.start()  # pylint: disable=no-member
+            self.pull_ufora_announcements.start()
+            self.remove_old_ufora_announcements.start()
 
     @tasks.loop(minutes=10)
     async def pull_ufora_announcements(self):
@@ -43,6 +44,15 @@ class Tasks(commands.Cog):
         """Error handler for the Ufora Announcements task"""
         print("".join(traceback.format_exception(type(error), error, error.__traceback__)))
 
+    @tasks.loop(hours=24)
+    async def remove_old_ufora_announcements(self):
+        """Remove all announcements that are over 1 week old, once per day"""
+        await remove_old_announcements(self.client.db_session)
+
+    @remove_old_ufora_announcements.before_loop
+    async def _before_remove_old_ufora_announcements(self):
+        await self.client.wait_until_ready()
+
 
 async def setup(client: Didier):
     """Load the cog"""