Feat: cache feed and update it after 10 minutes

.gitignore

@@ -1,2 +1,3 @@
 venv/
-__pycache__/
+__pycache__/
+database.db

database/feed.py

@@ -0,0 +1,28 @@
+from typing import Optional
+from sqlmodel import Field, SQLModel
+import datetime
+
+
+class FeedBase(SQLModel):
+    handle: str
+    feed: str
+
+
+class Feed(FeedBase, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    updated_at: Optional[datetime.datetime] = Field(default_factory=lambda: datetime.datetime.now())
+
+
+class FeedPublic(FeedBase):
+    id: int
+    updated_at: Optional[datetime.datetime]
+
+
+class FeedCreate(FeedBase):
+    pass
+
+
+class FeedUpdate(FeedBase):
+    handle: str | None = None
+    feed: str | None = None
+    updated_at: Optional[datetime.datetime]

main.py

@@ -1,11 +1,30 @@
-from fastapi import FastAPI, HTTPException, Response
+from fastapi import FastAPI, HTTPException, Response, Depends
+from sqlmodel import Session, select
+from typing import Annotated
+import datetime
+
+from database.feed import Feed, FeedCreate, FeedPublic, FeedUpdate
 from utils.feed_generator import generate_feed_of_user, USER_NOT_FOUND, CANNOT_ACCESS_INSTANCE, INVALID_HANDLE
+from utils.database import get_session, create_db_and_tables
 
 app = FastAPI()
+SessionDep = Annotated[Session, Depends(get_session)]
+
+
+@app.on_event("startup")
+def on_startup():
+    create_db_and_tables()
 
 
 @app.get("/feed/{user_handle}")
-def get_feed_of_user(user_handle: str):
+def get_feed_of_user(user_handle: str, session: SessionDep):
+    # get feed on database
+    feed_db = session.exec(select(Feed).where(Feed.handle == user_handle)).first()
+    
+    # return cached feed if it has been cached for less than 10 minutes
+    if feed_db and ((feed_db.updated_at + datetime.timedelta(minutes=10)) - datetime.datetime.now()).total_seconds() > 0:
+        return Response(content=feed_db.feed, media_type="application/xml")
+
     feed = generate_feed_of_user(user_handle)
 
     if feed == USER_NOT_FOUND:
@@ -15,4 +34,14 @@ def get_feed_of_user(user_handle: str):
     if feed == INVALID_HANDLE:
         return HTTPException(status_code=400, detail="The handle is invalid.")
     
+    # cache new feed
+    if feed_db:
+        feed_data = FeedUpdate(feed=feed, updated_at=datetime.datetime.now()).model_dump(exclude_unset=True)
+        feed_db.sqlmodel_update(feed_data)
+    else:
+        feed_db = Feed.model_validate(FeedCreate(handle=user_handle, feed=feed))
+    session.add(feed_db)
+    session.commit()
+    session.refresh(feed_db)
+    
     return Response(content=feed, media_type="application/xml")

requirements.txt

@@ -1,3 +1,4 @@
 fastapi[standard]==0.115.4
 requests==2.32.3
-rfeed==1.1.1
+rfeed==1.1.1
+sqlmodel==0.0.22

utils/database.py

@@ -0,0 +1,15 @@
+from sqlmodel import create_engine, Session, SQLModel
+
+sqlite_file_name = "database.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+connect_args = {"check_same_thread": False}
+engine = create_engine(sqlite_url, connect_args=connect_args)
+
+
+def create_db_and_tables():
+    SQLModel.metadata.create_all(engine)
+
+
+def get_session():
+    with Session(engine) as session:
+        yield session