mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-11 07:42:20 -05:00
267 lines
9.1 KiB
Python
267 lines
9.1 KiB
Python
from flask import Blueprint, current_app, jsonify, request
|
||
|
||
from app import api_user, authenticated_service
|
||
from app.aws.s3 import get_personalisation_from_s3, get_phone_number_from_s3
|
||
from app.dao import notifications_dao
|
||
from app.enums import KeyType, NotificationType
|
||
from app.errors import InvalidRequest, register_errors
|
||
from app.notifications.process_notifications import (
|
||
persist_notification,
|
||
send_notification_to_queue,
|
||
simulated_recipient,
|
||
)
|
||
from app.notifications.validators import (
|
||
check_if_service_can_send_to_number,
|
||
service_has_permission,
|
||
validate_template,
|
||
)
|
||
from app.public_schemas.public import PublicNotificationResponseSchema
|
||
from app.schemas import (
|
||
email_notification_schema,
|
||
notifications_filter_schema,
|
||
sms_template_notification_schema,
|
||
)
|
||
from app.service.utils import service_allowed_to_send_to
|
||
from app.utils import (
|
||
get_public_notify_type_text,
|
||
get_template_instance,
|
||
pagination_links,
|
||
template_model_to_dict,
|
||
)
|
||
from notifications_utils import SMS_CHAR_COUNT_LIMIT
|
||
|
||
notifications = Blueprint("notifications", __name__)
|
||
register_errors(notifications)
|
||
|
||
|
||
@notifications.route("/notifications/<uuid:notification_id>", methods=["GET"])
|
||
def get_notification_by_id(notification_id):
|
||
notification = notifications_dao.get_notification_with_personalisation(
|
||
str(authenticated_service.id), notification_id, key_type=None
|
||
)
|
||
|
||
if notification.job_id is not None:
|
||
notification.personalisation = get_personalisation_from_s3(
|
||
notification.service_id,
|
||
notification.job_id,
|
||
notification.job_row_number,
|
||
)
|
||
recipient = get_phone_number_from_s3(
|
||
notification.service_id,
|
||
notification.job_id,
|
||
notification.job_row_number,
|
||
)
|
||
notification.to = recipient
|
||
notification.normalised_to = recipient
|
||
|
||
if not hasattr(notification, "body") or not notification.body:
|
||
template_dict = template_model_to_dict(notification.template)
|
||
template = get_template_instance(
|
||
template_dict, notification.personalisation or {}
|
||
)
|
||
notification.body = template.content_with_placeholders_filled_in
|
||
if hasattr(template, "subject"):
|
||
try:
|
||
notification.subject = template.subject
|
||
except Exception:
|
||
# Subject attribute may not exist on notification model
|
||
pass
|
||
|
||
schema = PublicNotificationResponseSchema()
|
||
schema.context = {"notification_instance": notification}
|
||
serialized = schema.dump(notification)
|
||
|
||
return jsonify(data={"notification": serialized}), 200
|
||
|
||
|
||
@notifications.route("/notifications", methods=["GET"])
|
||
def get_all_notifications():
|
||
current_app.logger.debug("enter get_all_notifications()")
|
||
data = notifications_filter_schema.load(request.args)
|
||
current_app.logger.debug(
|
||
f"get_all_notifications() data {data} request.args {request.args}"
|
||
)
|
||
|
||
include_jobs = data.get("include_jobs", False)
|
||
page = data.get("page", 1)
|
||
page_size = data.get("page_size", current_app.config.get("API_PAGE_SIZE"))
|
||
limit_days = data.get("limit_days")
|
||
|
||
pagination = notifications_dao.get_notifications_for_service(
|
||
str(authenticated_service.id),
|
||
personalisation=True,
|
||
filter_dict=data,
|
||
page=page,
|
||
page_size=page_size,
|
||
limit_days=limit_days,
|
||
key_type=api_user.key_type,
|
||
include_jobs=include_jobs,
|
||
)
|
||
|
||
serialized = []
|
||
for notification in pagination.items:
|
||
personalisation = notification.personalisation
|
||
|
||
if notification.job_id is not None:
|
||
personalisation = get_personalisation_from_s3(
|
||
notification.service_id,
|
||
notification.job_id,
|
||
notification.job_row_number,
|
||
)
|
||
recipient = get_phone_number_from_s3(
|
||
notification.service_id,
|
||
notification.job_id,
|
||
notification.job_row_number,
|
||
)
|
||
# Safe to set dynamically for serialization purposes
|
||
notification.to = recipient
|
||
notification.normalised_to = recipient
|
||
|
||
subject = None
|
||
|
||
if not getattr(notification, "body", None):
|
||
template_dict = template_model_to_dict(notification.template)
|
||
template = get_template_instance(template_dict, personalisation or {})
|
||
notification.body = template.content_with_placeholders_filled_in
|
||
if hasattr(template, "subject"):
|
||
subject = template.subject
|
||
|
||
notification.personalisation = personalisation
|
||
|
||
schema = PublicNotificationResponseSchema()
|
||
schema.context = {"notification_instance": notification}
|
||
notification_data = schema.dump(notification)
|
||
|
||
if subject is not None:
|
||
notification_data["subject"] = subject
|
||
|
||
serialized.append(notification_data)
|
||
|
||
result = jsonify(
|
||
notifications=serialized,
|
||
page_size=page_size,
|
||
total=pagination.total,
|
||
links=pagination_links(
|
||
pagination, ".get_all_notifications", **request.args.to_dict()
|
||
),
|
||
)
|
||
return result, 200
|
||
|
||
|
||
@notifications.route("/notifications/<string:notification_type>", methods=["POST"])
|
||
def send_notification(notification_type):
|
||
if notification_type not in {NotificationType.SMS, NotificationType.EMAIL}:
|
||
raise InvalidRequest(
|
||
f"{notification_type} notification type is not supported", 400
|
||
)
|
||
|
||
notification_form = (
|
||
sms_template_notification_schema
|
||
if notification_type == NotificationType.SMS
|
||
else email_notification_schema
|
||
).load(request.get_json())
|
||
|
||
template, template_with_content = validate_template(
|
||
template_id=notification_form["template"],
|
||
personalisation=notification_form.get("personalisation", {}),
|
||
service=authenticated_service,
|
||
notification_type=notification_type,
|
||
)
|
||
|
||
_service_allowed_to_send_to(notification_form, authenticated_service)
|
||
|
||
if not service_has_permission(notification_type, authenticated_service.permissions):
|
||
raise InvalidRequest(
|
||
{
|
||
"service": [
|
||
f"Cannot send {get_public_notify_type_text(notification_type, plural=True)}"
|
||
]
|
||
},
|
||
400,
|
||
)
|
||
|
||
if notification_type == NotificationType.SMS:
|
||
check_if_service_can_send_to_number(
|
||
authenticated_service, notification_form["to"]
|
||
)
|
||
|
||
simulated = simulated_recipient(notification_form["to"], notification_type)
|
||
|
||
notification_model = persist_notification(
|
||
template_id=template.id,
|
||
template_version=template.version,
|
||
recipient=notification_form["to"],
|
||
service=authenticated_service,
|
||
personalisation=notification_form.get("personalisation"),
|
||
notification_type=notification_type,
|
||
api_key_id=api_user.id,
|
||
key_type=api_user.key_type,
|
||
simulated=simulated,
|
||
reply_to_text=template.reply_to_text,
|
||
)
|
||
|
||
if not simulated:
|
||
send_notification_to_queue(notification=notification_model, queue=None)
|
||
else:
|
||
current_app.logger.debug(
|
||
f"POST simulated notification for id: {notification_model.id}"
|
||
)
|
||
|
||
notification_form.update({"template_version": template.version})
|
||
|
||
return (
|
||
jsonify(
|
||
data=get_notification_return_data(
|
||
notification_model.id, notification_form, template_with_content
|
||
)
|
||
),
|
||
201,
|
||
)
|
||
|
||
|
||
def get_notification_return_data(notification_id, notification, template):
|
||
output = {
|
||
"template_version": notification["template_version"],
|
||
"notification": {"id": notification_id},
|
||
"body": template.content_with_placeholders_filled_in,
|
||
}
|
||
if hasattr(template, "subject"):
|
||
output["subject"] = template.subject
|
||
return output
|
||
|
||
|
||
def _service_allowed_to_send_to(notification, service):
|
||
if not service_allowed_to_send_to(notification["to"], service, api_user.key_type):
|
||
if api_user.key_type == KeyType.TEAM:
|
||
message = "Can’t send to this recipient using a team-only API key"
|
||
else:
|
||
message = (
|
||
"Can’t send to this recipient when service is in trial mode "
|
||
"– see https://www.notifications.service.gov.uk/trial-mode"
|
||
)
|
||
raise InvalidRequest({"to": [message]}, 400)
|
||
|
||
|
||
def create_template_object_for_notification(template, personalisation):
|
||
template_object = template._as_utils_template_with_personalisation(personalisation)
|
||
|
||
if template_object.missing_data:
|
||
message = "Missing personalisation: {}".format(
|
||
", ".join(template_object.missing_data)
|
||
)
|
||
raise InvalidRequest({"template": [message]}, 400)
|
||
|
||
if (
|
||
template_object.template_type == NotificationType.SMS
|
||
and template_object.is_message_too_long()
|
||
):
|
||
raise InvalidRequest(
|
||
{
|
||
"content": [
|
||
f"Content has a character count greater than the limit of {SMS_CHAR_COUNT_LIMIT}"
|
||
]
|
||
},
|
||
400,
|
||
)
|
||
return template_object
|