Refactor logic for injecting/creating RSS routes

This commit removes the duplicated code used to inject the logic
for templating RSS feeds into existing routes.

src/helpers/render.py

@@ -0,0 +1,48 @@
+import datetime
+from typing import Optional, Dict, Any
+
+import sanic
+import sanic_ext
+
+# Wrapper around sanic_ext.render
+def render_template(
+    template: str = "",
+    context: Optional[Dict[str, Any]] = None,
+    **kwargs
+):
+    jinja_context = context or {}
+
+    request = sanic.Request.get_current()
+
+    if hasattr(request.route.ctx, "rss"):
+        template = getattr(request.route.ctx, "template", None) or template
+        template = f"rss/{template}.xml"
+        kwargs["content_type"] = "application/rss+xml"
+
+        # Remove /rss suffix
+        base_path = request.app.url_for(request.endpoint[:-4], **request.match_info)
+
+        page_url = f"{request.app.ctx.PRIVIBLUR_CONFIG.deployment.domain or ''}{base_path}"
+        if request.query_string:
+            page_url += f"?{request.query_string}"
+
+        jinja_context["page_url"] = page_url
+
+        if (elements := jinja_context.get("blog")) and elements.posts:
+            jinja_context["updated"] = elements.posts[-1].date
+        elif (elements := jinja_context.get("timeline")) and elements.elements:
+            jinja_context["updated"] = elements.elements[-1].date
+        else:
+            jinja_context["updated"] = datetime.datetime.now(tz=datetime.timezone.utc)
+
+    template = f"{template}.jinja"
+
+    return sanic_ext.render(
+        template,
+        context=jinja_context,
+        app=request.app,
+        **kwargs
+    )
+
+
+

src/routes/blogs/blogs.py

@@ -1,8 +1,6 @@
-import datetime
 import urllib.parse
 
 import sanic
-import sanic_ext
 
 from ...cache import get_blog_posts, get_blog_search_results
 
@@ -22,28 +20,12 @@ async def _blog_posts(request: sanic.Request, blog: str):
 
     blog = await get_blog_posts(request.app.ctx, blog, continuation=continuation, before_id=before_id)
 
-    if hasattr(request.route.ctx, "rss"):
-        template_path = "rss/blog/blog.xml.jinja"
-        render_args : dict = {"content_type": "application/rss+xml"}
-
-        context_args : dict = {}
-        if last_post := blog.posts[-1]:
-            context_args["updated"] = last_post.date
-        else:
-            context_args["updated"] = datetime.datetime.now(tz=datetime.timezone.utc)
-    else:
-        template_path = "blog/blog.jinja"
-        render_args : dict = {}
-        context_args : dict = {}
-
-    return await sanic_ext.render(
-        template_path,
+    return await request.app.ctx.render(
+        "blog/blog",
         context={
             "app": request.app,
             "blog": blog,
-            **context_args
         },
-        **render_args
     )
 
 
@@ -60,29 +42,13 @@ async def _blog_tags(request: sanic.Request, blog: str, tag: str):
 
     blog = await get_blog_posts(request.app.ctx, blog, continuation=continuation, tag=tag)
 
-    if hasattr(request.route.ctx, "rss"):
-        template_path = "rss/blog/blog.xml.jinja"
-        render_args : dict = {"content_type": "application/rss+xml"}
-
-        context_args : dict = {}
-        if last_post := blog.posts[-1]:
-            context_args["updated"] = last_post.date
-        else:
-            context_args["updated"] = datetime.datetime.now(tz=datetime.timezone.utc)
-    else:
-        template_path = "blog/blog.jinja"
-        context_args : dict = {}
-        render_args : dict = {}
-
-    return await sanic_ext.render(
-        template_path,
+    return await request.app.ctx.render(
+        "blog/blog",
         context={
             "app": request.app,
             "blog": blog,
             "tag": tag,
-            **context_args
         },
-        **render_args
     )
 
 
@@ -99,29 +65,13 @@ async def _blog_search(request: sanic.Request, blog: str, query: str):
 
     blog = (await get_blog_search_results(request.app.ctx, blog, query, continuation=continuation))
 
-    if hasattr(request.route.ctx, "rss"):
-        template_path = "rss/blog/blog.xml.jinja"
-        render_args : dict = {"content_type": "application/rss+xml"}
-
-        context_args : dict = {}
-        if last_post := blog.posts[-1]:
-            context_args["updated"] = last_post.date
-        else:
-            context_args["updated"] = datetime.datetime.now(tz=datetime.timezone.utc)
-    else:
-        template_path = "blog/blog_search.jinja"
-        context_args : dict = {}
-        render_args : dict = {}
-
-    return await sanic_ext.render(
-        template_path,
+    return await request.app.ctx.render(
+        "blog/blog_search",
         context={
             "app": request.app,
             "blog": blog,
             "blog_search_query": query,
-            **context_args
         },
-        **render_args
     )
 
 

src/routes/explore.py

@@ -1,10 +1,7 @@
-import datetime
 import urllib.parse
 
 import sanic
-import sanic_ext
 
-from .. import priviblur_extractor
 from ..cache import get_explore_results
 
 explore = sanic.Blueprint("explore", url_prefix="/explore")
@@ -46,33 +43,13 @@ async def _handle_explore(request, endpoint, post_type = None):
             )
             title = request.app.ctx.translate(request.ctx.language, "explore_trending_page_title")
 
-    if hasattr(request.route.ctx, "rss"):
-        template_path = "rss/timeline.xml.jinja"
-        render_args : dict = {
-            "content_type": "application/rss+xml",
-        }
-
-        context_args : dict = {
-            "page_url": f"{request.app.ctx.PRIVIBLUR_CONFIG.deployment.domain or ''}/{endpoint}"
-        }
-        if last_post := timeline.elements[-1]:
-            context_args["updated"] = last_post.date
-        else:
-            context_args["updated"] = datetime.datetime.now(tz=datetime.timezone.utc)
-    else:
-        template_path = "timeline.jinja"
-        context_args : dict = {}
-        render_args : dict = {}
-
-    return await sanic_ext.render(
-        template_path,
+    return await request.app.ctx.render(
+        "timeline",
         context={
             "app": app,
             "title": title,
             "timeline": timeline,
-            **context_args
         },
-        **render_args
     )
 
 

src/routes/search.py

@@ -1,11 +1,8 @@
-import datetime
 import urllib.parse
 
 import sanic
-import sanic_ext
 
 from ..cache import get_search_results
-from .. import priviblur_extractor
 
 search = sanic.Blueprint("search", url_prefix="/search")
 
@@ -20,7 +17,7 @@ async def query_param_redirect(request: sanic.Request):
 
 
 @search.get("/<query:str>")
-@search.get("/<query:str>/rss", name="_name_rss", ctx_rss=True)
+@search.get("/<query:str>/rss", name="_main_rss", ctx_rss=True, ctx_template="timeline")
 async def _main(request: sanic.Request, query: str):
     query = urllib.parse.unquote(query)
 
@@ -34,7 +31,7 @@ async def _main(request: sanic.Request, query: str):
 
 
 @search.get("/<query:str>/recent")
-@search.get("/<query:str>/recent/rss", name="_sort_by_search_rss", ctx_rss=True)
+@search.get("/<query:str>/recent/rss", name="_sort_by_search_rss", ctx_rss=True, ctx_template="timeline")
 async def _sort_by_search(request: sanic.Request, query: str):
     query = urllib.parse.unquote(query)
     time_filter = request.args.get("t")
@@ -49,13 +46,13 @@ async def _sort_by_search(request: sanic.Request, query: str):
 
 
 @search.get("/<query:str>/<post_filter:str>")
-@search.get("/<query:str>/<post_filter:str>/rss", name="_filter_by_search_rss", ctx_rss=True)
+@search.get("/<query:str>/<post_filter:str>/rss", name="_filter_by_search_rss", ctx_rss=True, ctx_template="timeline")
 async def _filter_by_search(request: sanic.Request, query: str, post_filter: str):
     return await _request_search_filter_post(request, query, post_filter, latest=False)
 
 
 @search.get("/<query:str>/recent/<post_filter:str>")
-@search.get("/<query:str>/recent/<post_filter:str>/rss", name="_sort_by_and_filter_search_rss", ctx_rss=True)
+@search.get("/<query:str>/recent/<post_filter:str>/rss", name="_sort_by_and_filter_search_rss", ctx_rss=True, ctx_template="timeline")
 async def _sort_by_and_filter_search(request: sanic.Request, query: str, post_filter: str):
     return await _request_search_filter_post(request, query, post_filter, latest=True)
 
@@ -114,6 +111,7 @@ async def _query_search(request, query, **kwargs):
 
     return await get_search_results(request.app.ctx, query, continuation, **kwargs)
 
+
 async def _render(request, timeline, query, **kwargs):
     # We remove the continuation parameter used to fetch this page as to ensure the current continuation parameter isn't
     # added when applying a search filter
@@ -129,28 +127,6 @@ async def _render(request, timeline, query, **kwargs):
 
     context.update(kwargs)
 
-    if hasattr(request.route.ctx, "rss"):
-        template_path = "rss/timeline.xml.jinja"
-        render_args : dict = {
-            "content_type": "application/rss+xml",
-        }
-        search_url = request.app.url_for(request.endpoint, query=urllib.parse.quote(query), **kwargs)
-        page_url = f"{request.app.ctx.PRIVIBLUR_CONFIG.deployment.domain or ''}/{search_url}"
-        if request.query_string:
-            page_url += f"?{request.query_string}"
-
-        context["page_url"] = page_url
-
-        if last_post := timeline.elements[-1]:
-            context["updated"] = last_post.date
-        else:
-            context["updated"] = datetime.datetime.now(tz=datetime.timezone.utc)
-    else:
-        template_path = "search.jinja"
-        render_args : dict = {}
-
-    return await sanic_ext.render(
-        template_path,
-        context=context,
-        **render_args
+    return await request.app.ctx.render(
+        "search", context=context,
     )

src/routes/tagged.py

@@ -1,16 +1,14 @@
 import urllib.parse
 
 import sanic
-import sanic_ext
 
 from ..cache import get_tag_browse_results
-from ..helpers import helpers
 
 tagged = sanic.Blueprint("tagged", url_prefix="/tagged")
 
 
 @tagged.get("/<tag:str>")
-@tagged.get("/<tag:str>/rss", name="_main_rss", ctx_rss=True)
+@tagged.get("/<tag:str>/rss", name="_main_rss", ctx_rss=True, ctx_template="timeline")
 async def _main(request: sanic.Request, tag: str):
     tag = urllib.parse.unquote(tag)
     sort_by = request.args.get("sort")
@@ -31,36 +29,13 @@ async def _main(request: sanic.Request, tag: str):
     if request.args.get("continuation"):
         del request.args["continuation"]
 
-    if hasattr(request.route.ctx, "rss"):
-        template_path = "rss/timeline.xml.jinja"
-        render_args : dict = {
-            "content_type": "application/rss+xml",
-        }
-        page_url = f"{request.app.ctx.PRIVIBLUR_CONFIG.deployment.domain or ''}/tagged/{urllib.parse.quote(tag)}"
-        if request.query_string:
-            page_url += f"?{request.query_string}"
-
-        context_args : dict = {
-            "page_url": page_url
-        }
-        if last_post := timeline.elements[-1]:
-            context_args["updated"] = last_post.date
-        else:
-            context_args["updated"] = datetime.datetime.now(tz=datetime.timezone.utc)
-    else:
-        template_path = "tagged.jinja"
-        context_args : dict = {}
-        render_args : dict = {}
-
-    return await sanic_ext.render(
-        template_path,
+    return await request.app.ctx.render(
+        "tagged",
         context={
             "app": request.app,
             "query_args": request.args,
             "timeline": timeline,
             "tag": tag,
             "sort_by": sort_by,
-            **context_args
         },
-        **render_args
     )

src/server.py

@@ -15,7 +15,7 @@
 from . import routes, priviblur_extractor, preferences
 from .exceptions import error_handlers
 from .config import load_config
-from .helpers import setup_logging, helpers, i18n, ext_npf_renderer
+from .helpers import setup_logging, helpers, render, i18n, ext_npf_renderer
 from .version import VERSION, CURRENT_COMMIT
 
 
@@ -126,6 +126,8 @@ def create_image_client(url, timeout):
     else:
         app.ctx.CacheDb = None
 
+    app.ctx.render = render.render_template
+
     # Add additional jinja filters and functions
 
     app.ext.environment.add_extension("jinja2.ext.do")