Update contents render

Dockerfile.dev

@@ -1,4 +1,4 @@
-FROM python:3.12 as base
+FROM python:3.12-slim as base
 
 ENV PYTHONUNBUFFERED True
 ENV APP_HOME /app
@@ -11,4 +11,4 @@ COPY . ./
 RUN apt-get update && pip install --upgrade pip
 RUN pip install --no-cache-dir -r requirements.txt --prefer-binary
 
-CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080", "--reload"]

services/push.py

@@ -4,10 +4,9 @@
 
 from config import get_settings
 from db.models import Subscription, Notification, UserProfile
-from db.subscriptions import (
-    delete_subscriptions, get_subscriptions, get_template, get_user
-)
-from utils import dt2idt
+from db.subscriptions import delete_subscriptions, get_subscriptions, get_user
+from utils import logger
+from utils.contents import make_contents
 
 
 def send(subscription: Subscription, payload: dict):
@@ -35,8 +34,10 @@ async def send_webpush(notification: Notification):
             send(subscription, payload)
         except WebPushException as ex:
             traceback.print_exc()
-            if ex.response is not None and ex.response.status_code in [404, 410]:
-                not_sent.append(subscription.endpoint)
+            if ex.response is not None:
+                logger.error(ex.response.json())
+                if ex.response.status_code in [404, 410]:
+                    not_sent.append(subscription.endpoint)
 
     return await delete_subscriptions(not_sent) if not_sent else None
 
@@ -58,36 +59,3 @@ async def make_payload(notification: Notification, user: UserProfile) -> dict:
         }
     }
     return payload
-
-
-async def make_contents(notification: Notification, user: UserProfile) -> tuple[str, str, str]:
-    params = make_params(notification, user)
-
-    title, body, url = "", "", ""
-    template = await get_template(notification.type, user.language)
-
-    if not template:
-        template = await get_template(notification.type, "en")
-
-    if not template:
-        return title, body, url
-
-    title = template.title.replace("{{ ", "{").replace(" }}", "}").format(**params)
-    body = template.body.replace("{{ ", "{").replace(" }}", "}").format(**params)
-    url = params["context_uri"] or params["object_uri"]
-
-    return title, body, url
-
-
-def make_params(notification: Notification, user: UserProfile):
-    params:  dict[str, str | None] = {
-        "timestamp": dt2idt(notification.updated).rfc3339format(),
-        "target_uri": notification.target.id,
-        "object_uri": notification.object.id,
-        "object_name": notification.object.name,
-        "context_uri": notification.context.id if notification.context else None,
-        "actor_uri": notification.actor.id if notification.actor else None,
-        "actor_name": notification.actor.name if notification.actor else "Unknown",
-        "target_name": user.displayname or "Unknown",
-    }
-    return params

tasks/webhooks.py

@@ -1,11 +1,8 @@
 import json
-import logging
 import requests
 
 from db.models import Notification
-
-
-logger = logging.getLogger("uvicorn.error")
+from utils import logger
 
 
 def send_notification_to_webhook(notification: Notification, webhook_url: str) -> None:

tests/conftest.py

@@ -192,6 +192,6 @@ def valid_push_template_payload():
         "description": "Test Description",
         "type": [""],
         "language": "en",
-        "title": "Test Title",
+        "title": "Test Title: {{object_name}}",
         "body": "Test Body: {{ object_uri }}",
     }

tests/services/push_test.py

@@ -3,11 +3,9 @@
 import pytest
 from unittest.mock import patch, MagicMock
 from pywebpush import WebPushException
-from services.push import make_contents, make_params, make_payload, send
-from db.models import PushTemplate, Subscription
-from services.push import send_webpush
-from db.models import Notification
 
+from db.models import Notification, Subscription
+from services.push import make_payload, send, send_webpush
 
 
 @pytest.fixture
@@ -157,84 +155,3 @@ def test_make_pyload(mock_get_settings, mock_make_contents, valid_notification_p
             }
         }
     }
-
-
-@patch("services.push.get_template")
-@patch("services.push.make_params")
-def test_make_contents(mock_make_params, mock_get_template, valid_notification_payload, valid_push_template_payload):
-    notification = Notification(**valid_notification_payload)
-    template = PushTemplate(**valid_push_template_payload)
-    mock_make_params.return_value = {
-        "target_uri": notification.target.id,
-        "context_uri": notification.context.id,
-        "object_uri": notification.object.id,
-    }
-    mock_get_template.return_value = template
-
-    user = MagicMock(language="en")
-    title, body, url = asyncio.run(make_contents(notification, user))
-
-    assert title == "Test Title"
-    assert body == f"Test Body: {notification.object.id}"
-    assert url == notification.context.id
-    mock_make_params.assert_called_once_with(notification, user)
-    mock_get_template.assert_called_once_with(notification.type, user.language)
-
-
-@patch("services.push.get_template")
-@patch("services.push.make_params")
-def test_make_contents_retry(mock_make_params, mock_get_template, valid_notification_payload, valid_push_template_payload):
-    notification = Notification(**valid_notification_payload)
-    template = PushTemplate(**valid_push_template_payload)
-    mock_make_params.return_value = {
-        "target_uri": notification.target.id,
-        "context_uri": notification.context.id,
-        "object_uri": notification.object.id,
-    }
-    mock_get_template.side_effect = [None, template]
-
-    user = MagicMock(language="ja")
-    title, body, url = asyncio.run(make_contents(notification, user))
-
-    assert title == "Test Title"
-    assert body == f"Test Body: {notification.object.id}"
-    assert url == notification.context.id
-    mock_make_params.assert_called_once_with(notification, user)
-    mock_get_template.assert_any_call(notification.type, user.language)
-    mock_get_template.assert_any_call(notification.type, "en")
-
-
-@patch("services.push.get_template")
-@patch("services.push.make_params")
-def test_make_contents_no_template(mock_make_params, mock_get_template, valid_notification_payload, valid_push_template_payload):
-    notification = Notification(**valid_notification_payload)
-    template = PushTemplate(**valid_push_template_payload)
-    mock_make_params.return_value = {
-        "target_uri": notification.target.id,
-        "context_uri": notification.context.id,
-        "object_uri": notification.object.id,
-    }
-    mock_get_template.return_value = None
-
-    user = MagicMock(language="en")
-    title, body, url = asyncio.run(make_contents(notification, user))
-
-    assert title == ""
-    assert body == ""
-    assert url == ""
-    mock_make_params.assert_called_with(notification, user)
-    mock_get_template.assert_called_with(notification.type, user.language)
-    mock_get_template.assert_called_with(notification.type, "en")
-
-
-def test_make_params(valid_notification_payload):
-    notification = Notification(**valid_notification_payload)
-    user = MagicMock(language="en", displayname="Test User")
-    params = make_params(notification, user)
-    assert params["timestamp"] is not None
-    assert params["target_uri"] == notification.target.id
-    assert params["object_uri"] == notification.object.id
-    assert params["context_uri"] == notification.context.id
-    assert params["actor_uri"] == notification.actor.id
-    assert params["actor_name"] == notification.actor.name
-    assert params["target_name"] == "Test User"

tests/utils/contents_test.py

@@ -0,0 +1,89 @@
+import asyncio
+import pytest
+from unittest.mock import patch, MagicMock
+
+from db.models import Notification, PushTemplate
+from utils.contents import make_contents, make_params
+
+
+@patch("utils.contents.get_template")
+@patch("utils.contents.make_params")
+def test_make_contents(mock_make_params, mock_get_template, valid_notification_payload, valid_push_template_payload):
+    notification = Notification(**valid_notification_payload)
+    template = PushTemplate(**valid_push_template_payload)
+    mock_make_params.return_value = {
+        "target_uri": notification.target.id,
+        "context_uri": notification.context.id,
+        "object_uri": notification.object.id,
+        "object_name": "Sample Object",
+    }
+    mock_get_template.return_value = template
+
+    user = MagicMock(language="en")
+    title, body, url = asyncio.run(make_contents(notification, user))
+
+    assert title == "Test Title: Sample Object"
+    assert body == f"Test Body: {notification.object.id}"
+    assert url == notification.context.id
+    mock_make_params.assert_called_once_with(notification, user)
+    mock_get_template.assert_called_once_with(notification.type, user.language)
+
+
+@patch("utils.contents.get_template")
+@patch("utils.contents.make_params")
+def test_make_contents_retry(mock_make_params, mock_get_template, valid_notification_payload, valid_push_template_payload):
+    notification = Notification(**valid_notification_payload)
+    template = PushTemplate(**valid_push_template_payload)
+    mock_make_params.return_value = {
+        "target_uri": notification.target.id,
+        "context_uri": notification.context.id,
+        "object_uri": notification.object.id,
+        "object_name": "Sample Object",
+    }
+    mock_get_template.side_effect = [None, template]
+
+    user = MagicMock(language="ja")
+    title, body, url = asyncio.run(make_contents(notification, user))
+
+    assert title == "Test Title: Sample Object"
+    assert body == f"Test Body: {notification.object.id}"
+    assert url == notification.context.id
+    mock_make_params.assert_called_once_with(notification, user)
+    mock_get_template.assert_any_call(notification.type, user.language)
+    mock_get_template.assert_any_call(notification.type, "en")
+
+
+@patch("utils.contents.get_template")
+@patch("utils.contents.make_params")
+def test_make_contents_no_template(mock_make_params, mock_get_template, valid_notification_payload, valid_push_template_payload):
+    notification = Notification(**valid_notification_payload)
+    template = PushTemplate(**valid_push_template_payload)
+    mock_make_params.return_value = {
+        "target_uri": notification.target.id,
+        "context_uri": notification.context.id,
+        "object_uri": notification.object.id,
+    }
+    mock_get_template.return_value = None
+
+    user = MagicMock(language="en")
+    title, body, url = asyncio.run(make_contents(notification, user))
+
+    assert title == ""
+    assert body == ""
+    assert url == ""
+    mock_make_params.assert_called_with(notification, user)
+    mock_get_template.assert_called_with(notification.type, user.language)
+    mock_get_template.assert_called_with(notification.type, "en")
+
+
+def test_make_params(valid_notification_payload):
+    notification = Notification(**valid_notification_payload)
+    user = MagicMock(language="en", displayname="Test User")
+    params = make_params(notification, user)
+    assert params["timestamp"] is not None
+    assert params["target_uri"] == notification.target.id
+    assert params["object_uri"] == notification.object.id
+    assert params["context_uri"] == notification.context.id
+    assert params["actor_uri"] == notification.actor.id
+    assert params["actor_name"] == notification.actor.name
+    assert params["target_name"] == "Test User"

tests/utils/utils_test.py → tests/utils/datetimee_test.py

@@ -1,6 +1,6 @@
 import pytest
 from datetime import datetime, timedelta, timezone
-from utils import InboxDatetime, datetime_to_inboxdatetime
+from utils.datetime import InboxDatetime, datetime_to_inboxdatetime
 
 
 def test_inboxdatetime_rfc3339format():

utils/__init__.py

@@ -1,26 +1,6 @@
 import logging
-from datetime import datetime
 
+from .datetime import datetime_to_inboxdatetime
 
 logger = logging.getLogger("uvicorn.error")
-
-
-class InboxDatetime(datetime):
-
-    def rfc3339format(self):
-        """Return the current time in RFC3339 format.
-
-        Example:
-        >>> InboxDatetime.now(timezone.utc).rfc3339format()
-        "2025-03-05T14:00:00Z"
-        >>> InboxDatetime.now(ZoneInfo("Asia/Tokyo")).rfc3339format()
-        "2025-03-05T23:00:00+09:00"
-        """
-        return self.isoformat(timespec="seconds").replace("+00:00", "Z")
-
-
-def datetime_to_inboxdatetime(dt: datetime):
-    return InboxDatetime.fromisoformat(dt.isoformat())
-
-
 dt2idt = datetime_to_inboxdatetime

utils/contents.py

@@ -0,0 +1,47 @@
+import re
+
+from db.models import Notification, UserProfile
+from db.subscriptions import get_template
+from utils import dt2idt
+
+
+async def make_contents(
+    notification: Notification, user: UserProfile
+) -> tuple[str, str, str]:
+    params = make_params(notification, user)
+
+    title, body, url = "", "", ""
+    template = await get_template(notification.type, user.language)
+
+    if not template:
+        template = await get_template(notification.type, "en")
+
+    if not template:
+        return title, body, url
+
+    title = render(template.title, params)
+    body = render(template.body, params)
+    url = params["context_uri"] or params["object_uri"]
+
+    return title, body, url
+
+
+def render(template: str, data: dict):
+    pattern = r"\{\{\s*(\w+)\s*\}\}"
+    return re.sub(pattern, lambda match: str(data.get(match.group(1), "")), template)
+
+
+def make_params(
+    notification: Notification, user: UserProfile
+) -> dict[str, str | None]:
+    params = {
+        "timestamp": dt2idt(notification.updated).rfc3339format(),
+        "target_uri": notification.target.id,
+        "object_uri": notification.object.id,
+        "object_name": notification.object.name,
+        "context_uri": notification.context.id if notification.context else None,
+        "actor_uri": notification.actor.id if notification.actor else None,
+        "actor_name": notification.actor.name if notification.actor else "Unknown",
+        "target_name": user.displayname or "Unknown",
+    }
+    return params

utils/datetime.py

@@ -0,0 +1,19 @@
+from datetime import datetime
+
+
+class InboxDatetime(datetime):
+
+    def rfc3339format(self):
+        """Return the current time in RFC3339 format.
+
+        Example:
+        >>> InboxDatetime.now(timezone.utc).rfc3339format()
+        "2025-03-05T14:00:00Z"
+        >>> InboxDatetime.now(ZoneInfo("Asia/Tokyo")).rfc3339format()
+        "2025-03-05T23:00:00+09:00"
+        """
+        return self.isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def datetime_to_inboxdatetime(dt: datetime):
+    return InboxDatetime.fromisoformat(dt.isoformat())