darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-18 05:28:49 -04:00
Code Issues Packages Projects Releases Wiki Activity

notify-api-412 use black to enforce python style standards

Browse Source
This commit is contained in:
Kenneth Kehl
2023-08-23 10:35:43 -07:00
parent a7898118d7
commit 026dc14021
586 changed files with 33990 additions and 23461 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
app/notifications/rest.py
View File

@@ -26,28 +26,36 @@ from app.schemas import (
from app.service.utils import service_allowed_to_send_to
from app.utils import get_public_notify_type_text, pagination_links
notifications = Blueprint('notifications', __name__)
notifications = Blueprint("notifications", __name__)
register_errors(notifications)
@notifications.route('/notifications/<uuid:notification_id>', methods=['GET'])
@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)
return jsonify(data={"notification": notification_with_personalisation_schema.dump(notification)}), 200
str(authenticated_service.id), notification_id, key_type=None
)
return (
jsonify(
data={
"notification": notification_with_personalisation_schema.dump(
notification
)
}
),
200,
)
@notifications.route('/notifications', methods=['GET'])
@notifications.route("/notifications", methods=["GET"])
def get_all_notifications():
data = notifications_filter_schema.load(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')
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),
@@ -57,120 +65,141 @@ def get_all_notifications():
page_size=page_size,
limit_days=limit_days,
key_type=api_user.key_type,
include_jobs=include_jobs)
return jsonify(
notifications=notification_with_personalisation_schema.dump(pagination.items, many=True),
page_size=page_size,
total=pagination.total,
links=pagination_links(
pagination,
'.get_all_notifications',
**request.args.to_dict()
)
), 200
include_jobs=include_jobs,
)
return (
jsonify(
notifications=notification_with_personalisation_schema.dump(
pagination.items, many=True
),
page_size=page_size,
total=pagination.total,
links=pagination_links(
pagination, ".get_all_notifications", **request.args.to_dict()
),
),
200,
)
@notifications.route('/notifications/<string:notification_type>', methods=['POST'])
@notifications.route("/notifications/<string:notification_type>", methods=["POST"])
def send_notification(notification_type):
if notification_type not in [SMS_TYPE, EMAIL_TYPE]:
msg = "{} notification type is not supported".format(notification_type)
raise InvalidRequest(msg, 400)
notification_form = (
sms_template_notification_schema if notification_type == SMS_TYPE else email_notification_schema
sms_template_notification_schema
if notification_type == SMS_TYPE
else email_notification_schema
).load(request.get_json())
check_rate_limiting(authenticated_service, api_user)
template, template_with_content = validate_template(
template_id=notification_form['template'],
personalisation=notification_form.get('personalisation', {}),
template_id=notification_form["template"],
personalisation=notification_form.get("personalisation", {}),
service=authenticated_service,
notification_type=notification_type
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': ["Cannot send {}".format(get_public_notify_type_text(notification_type, plural=True))]},
status_code=400
{
"service": [
"Cannot send {}".format(
get_public_notify_type_text(notification_type, plural=True)
)
]
},
status_code=400,
)
if notification_type == SMS_TYPE:
check_if_service_can_send_to_number(authenticated_service, notification_form['to'])
check_if_service_can_send_to_number(
authenticated_service, notification_form["to"]
)
# Do not persist or send notification to the queue if it is a simulated recipient
simulated = simulated_recipient(notification_form['to'], notification_type)
notification_model = persist_notification(template_id=template.id,
template_version=template.version,
recipient=request.get_json()['to'],
service=authenticated_service,
personalisation=notification_form.get('personalisation', None),
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,
)
simulated = simulated_recipient(notification_form["to"], notification_type)
notification_model = persist_notification(
template_id=template.id,
template_version=template.version,
recipient=request.get_json()["to"],
service=authenticated_service,
personalisation=notification_form.get("personalisation", None),
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:
queue_name = QueueNames.PRIORITY if template.process_type == PRIORITY else None
send_notification_to_queue(notification=notification_model,
research_mode=authenticated_service.research_mode,
queue=queue_name)
send_notification_to_queue(
notification=notification_model,
research_mode=authenticated_service.research_mode,
queue=queue_name,
)
else:
current_app.logger.debug("POST simulated notification for id: {}".format(notification_model.id))
current_app.logger.debug(
"POST simulated notification for id: {}".format(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
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,
"template_version": notification["template_version"],
"notification": {"id": notification_id},
"body": template.content_with_placeholders_filled_in,
}
if hasattr(template, 'subject'):
output['subject'] = template.subject
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 not service_allowed_to_send_to(notification["to"], service, api_user.key_type):
if api_user.key_type == KEY_TYPE_TEAM:
message = 'Can’t send to this recipient using a team-only API key'
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'
"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]},
status_code=400
)
raise InvalidRequest({"to": [message]}, status_code=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))
errors = {'template': [message]}
message = "Missing personalisation: {}".format(
", ".join(template_object.missing_data)
)
errors = {"template": [message]}
raise InvalidRequest(errors, status_code=400)
if (
template_object.template_type == SMS_TYPE and
template_object.is_message_too_long()
template_object.template_type == SMS_TYPE
and template_object.is_message_too_long()
):
message = 'Content has a character count greater than the limit of {}'.format(SMS_CHAR_COUNT_LIMIT)
errors = {'content': [message]}
message = "Content has a character count greater than the limit of {}".format(
SMS_CHAR_COUNT_LIMIT
)
errors = {"content": [message]}
raise InvalidRequest(errors, status_code=400)
return template_object