Ali Asadi
c7af6bb14b
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>
115 lines
4 KiB
Python
115 lines
4 KiB
Python
import logging
|
|
|
|
import requests
|
|
import time
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
from django.db import models
|
|
from httpx import Request
|
|
|
|
from utils.clients.gooyal_notifications_client import AuthenticatedClient
|
|
from utils.clients.gooyal_notifications_client.api.email import email_application_application_create
|
|
from utils.clients.gooyal_notifications_client.api.push import push_application_application_create
|
|
from utils.clients.gooyal_notifications_client.models import Email
|
|
from utils.clients.gooyal_notifications_client.models import PushMessage
|
|
|
|
# logging.basicConfig(
|
|
# format="%(levelname)s [%(asctime)s] %(name)s - %(message)s",
|
|
# datefmt="%Y-%m-%d %H:%M:%S",
|
|
# level=logging.DEBUG
|
|
# )
|
|
logger = logging.getLogger('ads.utils.notifications_client')
|
|
|
|
|
|
|
|
def login_as_client_credentials():
|
|
access_token = cache.get('ads_access_token')
|
|
# expires_in = self.auth_data.get('expires_in')
|
|
# created_at = self.auth_data.get('created_at')
|
|
#
|
|
# if access_token and expires_in > time.time() - created_at:
|
|
# return self.auth_data
|
|
if access_token:
|
|
logger.debug('token available in cache')
|
|
return access_token
|
|
|
|
logger.debug('no token available in cache')
|
|
|
|
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)
|
|
logger.debug(response.content)
|
|
|
|
auth_data = response.json()
|
|
if 'access_token' in auth_data:
|
|
auth_data['created_at'] = time.time()
|
|
cache.set('ads_access_token', auth_data, timeout=auth_data.get('expires_in') - 5)
|
|
|
|
return auth_data
|
|
|
|
|
|
def log_request(request: Request):
|
|
logger.debug(f'authorization: {request.headers["authorization"]}')
|
|
logger.debug(
|
|
f"Request event hook: {request.method} {request.url} {request.method} {request.headers} - Waiting for response")
|
|
|
|
|
|
def log_response(response):
|
|
request = response.request
|
|
logger.debug(f"Response event hook: {request.method} {request.url} - Status {response.status_code}")
|
|
# logger.debug(f"Response event hook: {response.read()}")
|
|
|
|
|
|
def get_notifications_client():
|
|
access_token = login_as_client_credentials()['access_token']
|
|
base_url = settings.NOTIFICATIONS_BASE_PUBLIC_URL
|
|
client = AuthenticatedClient(base_url=base_url, token=access_token,
|
|
httpx_args={"event_hooks": {"request": [log_request], "response": [log_response]}},
|
|
)
|
|
return client
|
|
|
|
|
|
def notifications_push_user(user_uuid, title, message, priority=5, extras=None, click_url=None):
|
|
# click_url is optional — most notifications aren't clickable. When set,
|
|
# the notifications service embeds it as the tap destination.
|
|
class Tmp():
|
|
def to_dict(self):
|
|
return {
|
|
'title': title,
|
|
'message': message,
|
|
'priority': priority,
|
|
'extras': extras,
|
|
'click_url': click_url,
|
|
}
|
|
|
|
push_data = Tmp()
|
|
client = get_notifications_client()
|
|
response: PushMessage = push_application_application_create.sync(client=client, body=push_data, user_uuid=user_uuid)
|
|
|
|
return response
|
|
|
|
|
|
def submit_event_in_queue(title=None, message=None):
|
|
try:
|
|
class Tmp():
|
|
def to_dict(self):
|
|
return {
|
|
"queue": settings.NOTIFICATIONS_EVENT_QUEUE,
|
|
"title": title,
|
|
"message": message
|
|
}
|
|
|
|
email_data = Tmp()
|
|
client = get_notifications_client()
|
|
response: Email = email_application_application_create.sync(client=client, body=email_data,
|
|
user_uuid=settings.NOTIFICATIONS_OPERATOR_USER_UUID)
|
|
|
|
return response
|
|
except Exception as e:
|
|
logger.exception(e)
|