mirror of
https://github.com/GSA/notifications-admin.git
synced 2025-12-09 14:45:00 -05:00
119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
import json
|
|
|
|
from app.extensions import redis_client
|
|
from app.notify_client import NotifyAdminAPIClient, _attach_current_user
|
|
|
|
|
|
class NotificationApiClient(NotifyAdminAPIClient):
|
|
def get_notifications_for_service(
|
|
self,
|
|
service_id,
|
|
job_id=None,
|
|
template_type=None,
|
|
status=None,
|
|
page=None,
|
|
page_size=None,
|
|
count_pages=None,
|
|
limit_days=None,
|
|
include_jobs=None,
|
|
include_from_test_key=None,
|
|
format_for_csv=None,
|
|
to=None,
|
|
include_one_off=None,
|
|
):
|
|
params = {
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"template_type": template_type,
|
|
"status": status,
|
|
"include_jobs": include_jobs,
|
|
"include_from_test_key": include_from_test_key,
|
|
"format_for_csv": format_for_csv,
|
|
"to": to,
|
|
"include_one_off": include_one_off,
|
|
"count_pages": count_pages,
|
|
}
|
|
|
|
params = {k: v for k, v in params.items() if v is not None}
|
|
|
|
# if `to` is set it is likely PII like an email address or mobile which
|
|
# we do not want in our logs, so we do a POST request instead of a GET
|
|
method = self.post if to else self.get
|
|
kwargs = {"data": params} if to else {"params": params}
|
|
|
|
if job_id:
|
|
return method(
|
|
url="/service/{}/job/{}/notifications".format(service_id, job_id),
|
|
**kwargs,
|
|
)
|
|
else:
|
|
if limit_days is not None:
|
|
params["limit_days"] = limit_days
|
|
return method(url="/service/{}/notifications".format(service_id), **kwargs)
|
|
|
|
def send_notification(
|
|
self, service_id, *, template_id, recipient, personalisation, sender_id
|
|
):
|
|
data = {
|
|
"template_id": template_id,
|
|
"to": recipient,
|
|
"personalisation": personalisation,
|
|
}
|
|
if sender_id:
|
|
data["sender_id"] = sender_id
|
|
data = _attach_current_user(data)
|
|
return self.post(
|
|
url="/service/{}/send-notification".format(service_id), data=data
|
|
)
|
|
|
|
def get_notification(self, service_id, notification_id):
|
|
return self.get(
|
|
url="/service/{}/notifications/{}".format(service_id, notification_id)
|
|
)
|
|
|
|
def get_api_notifications_for_service(self, service_id):
|
|
ret = self.get_notifications_for_service(
|
|
service_id,
|
|
include_jobs=False,
|
|
include_from_test_key=True,
|
|
include_one_off=False,
|
|
count_pages=False,
|
|
)
|
|
return ret
|
|
|
|
def update_notification_to_cancelled(self, service_id, notification_id):
|
|
return self.post(
|
|
url="/service/{}/notifications/{}/cancel".format(
|
|
service_id, notification_id
|
|
),
|
|
data={},
|
|
)
|
|
|
|
def get_notification_status_by_service(self, start_date, end_date):
|
|
return self.get(
|
|
url="service/monthly-data-by-service",
|
|
params={
|
|
"start_date": str(start_date),
|
|
"end_date": str(end_date),
|
|
},
|
|
)
|
|
|
|
def get_notification_count_for_job_id(self, *, service_id, job_id):
|
|
counts = redis_client.get(
|
|
f"notification-count-for-job-id-{service_id}-{job_id}"
|
|
)
|
|
if counts is not None:
|
|
return json.loads(counts.decode("utf-8"))
|
|
result = self.get(
|
|
url="/service/{}/job/{}/notification_count".format(service_id, job_id)
|
|
)
|
|
redis_client.set(
|
|
f"notification-count-for-job-id-{service_id}-{job_id}",
|
|
json.dumps(result["count"]),
|
|
ex=30,
|
|
)
|
|
return result["count"]
|
|
|
|
|
|
notification_api_client = NotificationApiClient()
|