Ali Asadi
780534a426
Passes through to the notifications service, which embeds it as the Gotify tap destination when set. Optional — no existing call site changes, most notifications stay non-clickable. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
import time
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
|
|
from utils.clients.gooyal_notifications_client import AuthenticatedClient
|
|
from utils.clients.gooyal_notifications_client.api.push import push_application_application_create
|
|
|
|
ACCESS_TOKEN_CACHE_KEY = "chat_notifications_access_token"
|
|
|
|
|
|
def login_as_client_credentials() -> dict:
|
|
"""
|
|
Client-credentials OAuth2 grant against the gooyal accounts service,
|
|
reusing a cached access token until it's within 5s of expiry.
|
|
"""
|
|
cached = cache.get(ACCESS_TOKEN_CACHE_KEY)
|
|
if cached:
|
|
return cached
|
|
|
|
data = {
|
|
"grant_type": "client_credentials",
|
|
"scope": settings.OAUTH2_PROVIDER_SCOPES,
|
|
}
|
|
auth = (settings.OAUTH2_PROVIDER_CLIENT_ID, settings.OAUTH2_PROVIDER_CLIENT_SECRET)
|
|
|
|
response = requests.post(
|
|
f"{settings.OAUTH2_PROVIDER_BASE_PUBLIC_URL}/token/", data=data, auth=auth
|
|
)
|
|
response.raise_for_status()
|
|
|
|
auth_data = response.json()
|
|
if "access_token" in auth_data:
|
|
auth_data["created_at"] = time.time()
|
|
cache.set(
|
|
ACCESS_TOKEN_CACHE_KEY, auth_data, timeout=auth_data.get("expires_in", 300) - 5
|
|
)
|
|
|
|
return auth_data
|
|
|
|
|
|
def get_client() -> AuthenticatedClient:
|
|
access_token = login_as_client_credentials()["access_token"]
|
|
return AuthenticatedClient(
|
|
base_url=settings.NOTIFICATIONS_BASE_PUBLIC_URL, token=access_token
|
|
)
|
|
|
|
|
|
class _PushBody:
|
|
"""
|
|
Duck-typed request body: push_user is already in the URL and
|
|
application is resolved server-side from the OAuth2 client, so only
|
|
these fields are sent (mirrors the advertising service's client).
|
|
"""
|
|
|
|
def __init__(self, title: str, message: str, priority: int, extras, click_url=None):
|
|
self._title = title
|
|
self._message = message
|
|
self._priority = priority
|
|
self._extras = extras
|
|
self._click_url = click_url
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"title": self._title,
|
|
"message": self._message,
|
|
"priority": self._priority,
|
|
"extras": self._extras,
|
|
"click_url": self._click_url,
|
|
}
|
|
|
|
|
|
def push_user(user_uuid, title: str, message: str, priority: int = 5, extras=None, click_url=None):
|
|
"""
|
|
Send a push notification to a single user via the gooyal notifications
|
|
service. Requires the `notifications.application.push:send` scope on
|
|
this app's OAuth2 client credentials. Raises on failure — callers
|
|
decide whether a down notifications service should be swallowed.
|
|
|
|
`click_url` is optional — most notifications aren't clickable. When set,
|
|
the notifications service embeds it as the tap destination.
|
|
"""
|
|
client = get_client()
|
|
body = _PushBody(title=title, message=message, priority=priority, extras=extras, click_url=click_url)
|
|
return push_application_application_create.sync(user_uuid=str(user_uuid), client=client, body=body)
|