From 7885f51b77630e9a65449e7fb138774d94dd334c Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Mon, 4 Aug 2025 14:04:54 -0700 Subject: [PATCH 1/8] try again --- app/celery/tasks.py | 116 ++++++++++++++++++++++++++++++++++++++++++++ app/service/rest.py | 68 ++++++++++++++++---------- 2 files changed, 158 insertions(+), 26 deletions(-) diff --git a/app/celery/tasks.py b/app/celery/tasks.py index 7cf3c0933..d31232155 100644 --- a/app/celery/tasks.py +++ b/app/celery/tasks.py @@ -33,6 +33,122 @@ from app.service.utils import service_allowed_to_send_to from app.utils import DATETIME_FORMAT, hilite, utc_now from notifications_utils.recipients import RecipientCSV +# @notify_celery.task(name="process-job") +# def generate_notifications_report(service_id, report_id, data, request_method): +# current_app.logger.debug(hilite("ENTER generate_notifications_report")) +# page = data["page"] if "page" in data else 1 +# page_size = ( +# data["page_size"] +# if "page_size" in data +# else current_app.config.get("PAGE_SIZE") +# ) +# # HARD CODE TO 100 for now. 1000 or 10000 causes reports to time out before they complete (if big) +# # Tests are relying on the value in config (20), whereas the UI seems to pass 10000 +# if page_size > 100: +# page_size = 100 +# limit_days = data.get("limit_days") +# include_jobs = data.get("include_jobs", True) +# include_from_test_key = data.get("include_from_test_key", False) +# include_one_off = data.get("include_one_off", True) + +# # count_pages is not being used for whether to count the number of pages, but instead as a flag +# # for whether to show pagination links +# count_pages = data.get("count_pages", True) + +# current_app.logger.debug( +# f"get pagination with {service_id} service_id filters {data} \ +# limit_days {limit_days} include_jobs {include_jobs} include_one_off {include_one_off}" +# ) +# start_time = time.time() +# current_app.logger.debug(f"Start report generation with page.size {page_size}") +# pagination = notifications_dao.get_notifications_for_service( +# service_id, +# filter_dict=data, +# page=page, +# page_size=page_size, +# count_pages=False, +# limit_days=limit_days, +# include_jobs=include_jobs, +# include_from_test_key=include_from_test_key, +# include_one_off=include_one_off, +# ) +# current_app.logger.debug(f"Query complete at {int(time.time()-start_time)*1000}") + +# for notification in pagination.items: +# if notification.job_id is not None: +# current_app.logger.debug( +# f"Processing job_id {notification.job_id} at {int(time.time()-start_time)*1000}" +# ) +# 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 + +# else: +# notification.to = "" +# notification.normalised_to = "" + +# kwargs = request.args.to_dict() +# kwargs["service_id"] = service_id + +# if data.get("format_for_csv"): +# notifications = [ +# notification.serialize_for_csv() for notification in pagination.items +# ] +# else: +# notifications = notification_with_template_schema.dump( +# pagination.items, many=True +# ) +# current_app.logger.debug(f"number of notifications are {len(notifications)}") + +# # We try and get the next page of results to work out if we need provide a pagination link to the next page +# # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous +# # call to be True which will enable us to use Flask-Sqlalchemy to tell if there is a next page of results but +# # this way is much more performant for services with many results (unlike Flask SqlAlchemy, this approach +# # doesn't do an additional query to count all the results of which there could be millions but instead only +# # asks for a single extra page of results). +# next_page_of_pagination = notifications_dao.get_notifications_for_service( +# service_id, +# filter_dict=data, +# page=page + 1, +# page_size=page_size, +# count_pages=False, +# limit_days=limit_days, +# include_jobs=include_jobs, +# include_from_test_key=include_from_test_key, +# include_one_off=include_one_off, +# error_out=False, # False so that if there are no results, it doesn't end in aborting with a 404 +# ) + +# x = ( +# jsonify( +# notifications=notifications, +# page_size=page_size, +# links=( +# get_prev_next_pagination_links( +# page, +# len(next_page_of_pagination.items), +# ".get_all_notifications_for_service", +# **kwargs, +# ) +# if count_pages +# else {} +# ), +# ), +# 200, +# ) +# current_app.logger.debug(x) + @notify_celery.task(name="process-job") def process_job(job_id, sender_id=None): diff --git a/app/service/rest.py b/app/service/rest.py index b204a940e..e1ed62ed8 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -1,5 +1,6 @@ import itertools import time +import uuid from datetime import datetime, timedelta from zoneinfo import ZoneInfo @@ -10,7 +11,7 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm.exc import NoResultFound from werkzeug.datastructures import MultiDict -from app import db +from app import db, notify_celery from app.aws.s3 import get_personalisation_from_s3, get_phone_number_from_s3 from app.config import QueueNames from app.dao import fact_notification_status_dao, notifications_dao @@ -116,7 +117,12 @@ from app.service.service_senders_schema import ( ) from app.service.utils import get_guest_list_objects from app.user.users_schema import post_set_permissions_schema -from app.utils import check_suspicious_id, get_prev_next_pagination_links, utc_now +from app.utils import ( + check_suspicious_id, + get_prev_next_pagination_links, + hilite, + utc_now, +) service_blueprint = Blueprint("service", __name__) @@ -511,27 +517,12 @@ def get_service_history(service_id): return jsonify(data=data) -@service_blueprint.route("//notifications", methods=["GET", "POST"]) -def get_all_notifications_for_service(service_id): - check_suspicious_id(service_id) - current_app.logger.debug("enter get_all_notifications_for_service") - if request.method == "GET": - data = notifications_filter_schema.load(request.args) - current_app.logger.debug( - f"use GET, request.args {request.args} and data {data}" - ) - elif request.method == "POST": - # Must transform request.get_json() to MultiDict as NotificationsFilterSchema expects a MultiDict. - # Unlike request.args, request.get_json() does not return a MultiDict but instead just a dict. - data = notifications_filter_schema.load(MultiDict(request.get_json())) - current_app.logger.debug(f"use POST, request {request.get_json()} data {data}") - +@notify_celery.task(name="generate-notifications-report") +def generate_notifications_report(service_id, report_id, data, request_method): + current_app.logger.debug(hilite("ENTER generate_notifications_report")) page = data["page"] if "page" in data else 1 - page_size = ( - data["page_size"] - if "page_size" in data - else current_app.config.get("PAGE_SIZE") - ) + page_size = data["page_size"] if "page_size" in data else 100 + current_app.logger.debug(hilite("PAGE SIZE 100")) # HARD CODE TO 100 for now. 1000 or 10000 causes reports to time out before they complete (if big) # Tests are relying on the value in config (20), whereas the UI seems to pass 10000 if page_size > 100: @@ -544,7 +535,7 @@ def get_all_notifications_for_service(service_id): # count_pages is not being used for whether to count the number of pages, but instead as a flag # for whether to show pagination links count_pages = data.get("count_pages", True) - + current_app.logger.debug(hilite("GET ALL PARAMS")) current_app.logger.debug( f"get pagination with {service_id} service_id filters {data} \ limit_days {limit_days} include_jobs {include_jobs} include_one_off {include_one_off}" @@ -591,7 +582,8 @@ def get_all_notifications_for_service(service_id): kwargs = request.args.to_dict() kwargs["service_id"] = service_id - if data.get("format_for_csv"): + if True: + # if data.get("format_for_csv"): notifications = [ notification.serialize_for_csv() for notification in pagination.items ] @@ -599,7 +591,9 @@ def get_all_notifications_for_service(service_id): notifications = notification_with_template_schema.dump( pagination.items, many=True ) - current_app.logger.debug(f"number of notifications are {len(notifications)}") + current_app.logger.debug( + hilite(f"number of notifications are {len(notifications)}") + ) # We try and get the next page of results to work out if we need provide a pagination link to the next page # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous @@ -607,6 +601,7 @@ def get_all_notifications_for_service(service_id): # this way is much more performant for services with many results (unlike Flask SqlAlchemy, this approach # doesn't do an additional query to count all the results of which there could be millions but instead only # asks for a single extra page of results). + current_app.logger.debug(hilite("ABOUT TO CALL NOTIFICATIONS DAO")) next_page_of_pagination = notifications_dao.get_notifications_for_service( service_id, filter_dict=data, @@ -619,8 +614,9 @@ def get_all_notifications_for_service(service_id): include_one_off=include_one_off, error_out=False, # False so that if there are no results, it doesn't end in aborting with a 404 ) + current_app.logger.debug(hilite("NOTIFICATIONS_DAO CALLED SUCCESSFULLY")) - return ( + x = ( jsonify( notifications=notifications, page_size=page_size, @@ -637,6 +633,26 @@ def get_all_notifications_for_service(service_id): ), 200, ) + current_app.logger.debug(hilite(x)) + + +@service_blueprint.route("//notifications", methods=["GET", "POST"]) +def get_all_notifications_for_service(service_id): + current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) + check_suspicious_id(service_id) + report_id = str(uuid.uuid4()) + request_method = request.method + if request_method == "GET": + data = notifications_filter_schema.load(request.args) + else: + data = notifications_filter_schema.load( + MultiDict(request.get_json(silent=True)) + ) + current_app.logger.debug(hilite("INVOKE APPLY_ASYNC")) + generate_notifications_report.apply_async( + args=[service_id, report_id, data, request_method], queue=QueueNames.NOTIFY + ) + return jsonify({"report_id": report_id}), 200 @service_blueprint.route( From 270189c00ee54e889ff132ebeaf010f83cd5b362 Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Tue, 5 Aug 2025 08:11:22 -0700 Subject: [PATCH 2/8] cleanup --- app/__init__.py | 3 +- app/service/rest.py | 197 +++++++++++++++++++++++++++++++++++++------- 2 files changed, 166 insertions(+), 34 deletions(-) diff --git a/app/__init__.py b/app/__init__.py index 05ac4c0e7..598c44313 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -480,8 +480,7 @@ def make_task(app): ) ) - def on_failure(self, exc, task_id, args, kwargs, einfo): - current_app.logger.debug(f"Using {einfo}") + def on_failure(self, exc, task_id, args, kwargs, einfo=None): # enables request id tracing for these logs with self.app_context(): diff --git a/app/service/rest.py b/app/service/rest.py index e1ed62ed8..87394fe3c 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -1,4 +1,8 @@ +import csv +import io import itertools +import logging +import os import time import uuid from datetime import datetime, timedelta @@ -123,11 +127,26 @@ from app.utils import ( hilite, utc_now, ) +from notifications_utils.s3 import s3upload + +celery_logger = logging.getLogger(__name__) service_blueprint = Blueprint("service", __name__) register_errors(service_blueprint) +NEW_FILE_LOCATION_STRUCTURE = "{}-service-notify/{}.csv" + + +def get_csv_location(service_id, upload_id): + return ( + current_app.config["CSV_UPLOAD_BUCKET"]["bucket"], + NEW_FILE_LOCATION_STRUCTURE.format(service_id, upload_id), + current_app.config["CSV_UPLOAD_BUCKET"]["access_key_id"], + current_app.config["CSV_UPLOAD_BUCKET"]["secret_access_key"], + current_app.config["CSV_UPLOAD_BUCKET"]["region"], + ) + @service_blueprint.errorhandler(IntegrityError) def handle_integrity_error(exc): @@ -518,11 +537,150 @@ def get_service_history(service_id): @notify_celery.task(name="generate-notifications-report") -def generate_notifications_report(service_id, report_id, data, request_method): +def generate_notifications_report( + service_id, report_id, data, request_method, request_args +): current_app.logger.debug(hilite("ENTER generate_notifications_report")) + page = 1 + page_size = 20000 + limit_days = data.get("limit_days") + include_jobs = data.get("include_jobs", True) + include_from_test_key = data.get("include_from_test_key", False) + include_one_off = data.get("include_one_off", True) + + current_app.logger.debug(hilite("GET ALL PARAMS")) + current_app.logger.debug( + f"get pagination with {service_id} service_id filters {data} \ + limit_days {limit_days} include_jobs {include_jobs} include_one_off {include_one_off}" + ) + start_time = time.time() + current_app.logger.debug(f"Start report generation with page.size {page_size}") + pagination = notifications_dao.get_notifications_for_service( + service_id, + filter_dict=data, + page=page, + page_size=page_size, + count_pages=False, + limit_days=limit_days, + include_jobs=include_jobs, + include_from_test_key=include_from_test_key, + include_one_off=include_one_off, + ) + current_app.logger.debug(f"Query complete at {int(time.time()-start_time)*1000}") + + for notification in pagination.items: + if notification.job_id is not None: + current_app.logger.debug( + f"Processing job_id {notification.job_id} at {int(time.time()-start_time)*1000}" + ) + 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 + + else: + notification.to = "" + notification.normalised_to = "" + + kwargs = request_args + kwargs["service_id"] = service_id + + notifications = [ + notification.serialize_for_csv() for notification in pagination.items + ] + current_app.logger.debug(hilite(f"NUMBER OF NOTFICIATIONS IS {len(notifications)}")) + + # We try and get the next page of results to work out if we need provide a pagination link to the next page + # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous + # call to be True which will enable us to use Flask-Sqlalchemy to tell if there is a next page of results but + # this way is much more performant for services with many results (unlike Flask SqlAlchemy, this approach + # doesn't do an additional query to count all the results of which there could be millions but instead only + # asks for a single extra page of results). + + csv_bytes = io.BytesIO() + text_wrapper = io.TextIOWrapper(csv_bytes, encoding="utf-8", newline="") + writer = csv.writer(text_wrapper) + writer.writerows(notifications) + text_wrapper.flush() + csv_bytes.seek(0) + + bucket_name, file_location, access_key, secret_key, region = get_csv_location( + service_id, report_id + ) + if bucket_name == "": + exp_bucket = current_app.config["CSV_UPLOAD_BUCKET"]["bucket"] + exp_region = current_app.config["CSV_UPLOAD_BUCKET"]["region"] + tier = os.getenv("NOTIFY_ENVIRONMENT") + raise Exception( + f"NO BUCKET NAME SHOULD BE: {exp_bucket} WITH REGION {exp_region} TIER {tier}" + ) + + current_app.logger.debug(f"UPLOADING THIS {csv_bytes}") + s3upload( + filedata=csv_bytes, + region=region, + bucket_name=bucket_name, + file_location=file_location, + access_key=access_key, + secret_key=secret_key, + ) + current_app.logger.debug(hilite("FINITO")) + + +@service_blueprint.route( + "//notifications-report", methods=["GET", "POST"] +) +def get_notifications_report_for_service(service_id): + current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) + check_suspicious_id(service_id) + report_id = str(uuid.uuid4()) + request_method = request.method + if request_method == "GET": + data = notifications_filter_schema.load(request.args) + else: + data = notifications_filter_schema.load( + MultiDict(request.get_json(silent=True)) + ) + request_args = request.args.to_dict() + current_app.logger.debug(hilite("INVOKE APPLY_ASYNC")) + generate_notifications_report.apply_async( + args=[service_id, report_id, data, request_method, request_args], + queue=QueueNames.NOTIFY, + ) + return jsonify({"report_id": report_id}), 200 + + +@service_blueprint.route("//notifications", methods=["GET", "POST"]) +def get_all_notifications_for_service(service_id): + check_suspicious_id(service_id) + current_app.logger.debug("enter get_all_notifications_for_service") + if request.method == "GET": + data = notifications_filter_schema.load(request.args) + current_app.logger.debug( + f"use GET, request.args {request.args} and data {data}" + ) + elif request.method == "POST": + # Must transform request.get_json() to MultiDict as NotificationsFilterSchema expects a MultiDict. + # Unlike request.args, request.get_json() does not return a MultiDict but instead just a dict. + data = notifications_filter_schema.load(MultiDict(request.get_json())) + current_app.logger.debug(f"use POST, request {request.get_json()} data {data}") + page = data["page"] if "page" in data else 1 - page_size = data["page_size"] if "page_size" in data else 100 - current_app.logger.debug(hilite("PAGE SIZE 100")) + page_size = ( + data["page_size"] + if "page_size" in data + else current_app.config.get("PAGE_SIZE") + ) # HARD CODE TO 100 for now. 1000 or 10000 causes reports to time out before they complete (if big) # Tests are relying on the value in config (20), whereas the UI seems to pass 10000 if page_size > 100: @@ -535,7 +693,7 @@ def generate_notifications_report(service_id, report_id, data, request_method): # count_pages is not being used for whether to count the number of pages, but instead as a flag # for whether to show pagination links count_pages = data.get("count_pages", True) - current_app.logger.debug(hilite("GET ALL PARAMS")) + current_app.logger.debug( f"get pagination with {service_id} service_id filters {data} \ limit_days {limit_days} include_jobs {include_jobs} include_one_off {include_one_off}" @@ -582,8 +740,7 @@ def generate_notifications_report(service_id, report_id, data, request_method): kwargs = request.args.to_dict() kwargs["service_id"] = service_id - if True: - # if data.get("format_for_csv"): + if data.get("format_for_csv"): notifications = [ notification.serialize_for_csv() for notification in pagination.items ] @@ -591,9 +748,7 @@ def generate_notifications_report(service_id, report_id, data, request_method): notifications = notification_with_template_schema.dump( pagination.items, many=True ) - current_app.logger.debug( - hilite(f"number of notifications are {len(notifications)}") - ) + current_app.logger.debug(f"number of notifications are {len(notifications)}") # We try and get the next page of results to work out if we need provide a pagination link to the next page # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous @@ -601,7 +756,6 @@ def generate_notifications_report(service_id, report_id, data, request_method): # this way is much more performant for services with many results (unlike Flask SqlAlchemy, this approach # doesn't do an additional query to count all the results of which there could be millions but instead only # asks for a single extra page of results). - current_app.logger.debug(hilite("ABOUT TO CALL NOTIFICATIONS DAO")) next_page_of_pagination = notifications_dao.get_notifications_for_service( service_id, filter_dict=data, @@ -614,9 +768,8 @@ def generate_notifications_report(service_id, report_id, data, request_method): include_one_off=include_one_off, error_out=False, # False so that if there are no results, it doesn't end in aborting with a 404 ) - current_app.logger.debug(hilite("NOTIFICATIONS_DAO CALLED SUCCESSFULLY")) - x = ( + return ( jsonify( notifications=notifications, page_size=page_size, @@ -633,26 +786,6 @@ def generate_notifications_report(service_id, report_id, data, request_method): ), 200, ) - current_app.logger.debug(hilite(x)) - - -@service_blueprint.route("//notifications", methods=["GET", "POST"]) -def get_all_notifications_for_service(service_id): - current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) - check_suspicious_id(service_id) - report_id = str(uuid.uuid4()) - request_method = request.method - if request_method == "GET": - data = notifications_filter_schema.load(request.args) - else: - data = notifications_filter_schema.load( - MultiDict(request.get_json(silent=True)) - ) - current_app.logger.debug(hilite("INVOKE APPLY_ASYNC")) - generate_notifications_report.apply_async( - args=[service_id, report_id, data, request_method], queue=QueueNames.NOTIFY - ) - return jsonify({"report_id": report_id}), 200 @service_blueprint.route( From 0e9760c2912f5f26fedbb1372122adb8a9b7cb9a Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Tue, 5 Aug 2025 09:01:41 -0700 Subject: [PATCH 3/8] clean up --- app/service/rest.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/app/service/rest.py b/app/service/rest.py index 87394fe3c..f479212d7 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -567,12 +567,14 @@ def generate_notifications_report( include_one_off=include_one_off, ) current_app.logger.debug(f"Query complete at {int(time.time()-start_time)*1000}") - + current_app.logger.debug(f"HOW MANY ITEMS IN PAGINATION? {len(pagination.items)}") + count = 1 for notification in pagination.items: if notification.job_id is not None: current_app.logger.debug( - f"Processing job_id {notification.job_id} at {int(time.time()-start_time)*1000}" + f"Processing job_id {notification.job_id} which is row {count}" ) + count = count + 1 notification.personalisation = get_personalisation_from_s3( notification.service_id, notification.job_id, @@ -638,7 +640,7 @@ def generate_notifications_report( @service_blueprint.route( - "//notifications-report", methods=["GET", "POST"] + "//notifications", methods=["GET", "POST"] ) def get_notifications_report_for_service(service_id): current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) @@ -660,7 +662,7 @@ def get_notifications_report_for_service(service_id): return jsonify({"report_id": report_id}), 200 -@service_blueprint.route("//notifications", methods=["GET", "POST"]) +@service_blueprint.route("//notificationsx", methods=["GET", "POST"]) def get_all_notifications_for_service(service_id): check_suspicious_id(service_id) current_app.logger.debug("enter get_all_notifications_for_service") From 1f1c97125734e42dac9987a80cf66d770cad9884 Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Tue, 5 Aug 2025 09:09:26 -0700 Subject: [PATCH 4/8] fix dead code warning --- app/__init__.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/app/__init__.py b/app/__init__.py index 598c44313..2a212288d 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -480,10 +480,11 @@ def make_task(app): ) ) - def on_failure(self, exc, task_id, args, kwargs, einfo=None): + def on_failure(self, exc, task_id, args, kwargs, einfo): # enables request id tracing for these logs with self.app_context(): + app.logger.debug(f"einfo is {einfo}") app.logger.exception( "Celery task {task_name} (queue: {queue_name}) failed".format( task_name=self.name, From bb68370eaee39a48ba06f80099ed275680520186 Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Tue, 5 Aug 2025 09:50:55 -0700 Subject: [PATCH 5/8] fix dead code warning --- app/service/rest.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/app/service/rest.py b/app/service/rest.py index f479212d7..c4d94e0d1 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -639,9 +639,7 @@ def generate_notifications_report( current_app.logger.debug(hilite("FINITO")) -@service_blueprint.route( - "//notifications", methods=["GET", "POST"] -) +@service_blueprint.route("//notifications-report", methods=["GET", "POST"]) def get_notifications_report_for_service(service_id): current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) check_suspicious_id(service_id) @@ -662,7 +660,7 @@ def get_notifications_report_for_service(service_id): return jsonify({"report_id": report_id}), 200 -@service_blueprint.route("//notificationsx", methods=["GET", "POST"]) +@service_blueprint.route("//notifications", methods=["GET", "POST"]) def get_all_notifications_for_service(service_id): check_suspicious_id(service_id) current_app.logger.debug("enter get_all_notifications_for_service") From d53e033d2af5539449759b58d355cda66e6fe8fb Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Tue, 5 Aug 2025 13:41:57 -0700 Subject: [PATCH 6/8] add task and command --- app/celery/tasks.py | 245 +++++++++++++++++++++++--------------------- app/commands.py | 10 +- app/config.py | 5 + app/service/rest.py | 146 +------------------------- 4 files changed, 144 insertions(+), 262 deletions(-) diff --git a/app/celery/tasks.py b/app/celery/tasks.py index d31232155..b665f2d09 100644 --- a/app/celery/tasks.py +++ b/app/celery/tasks.py @@ -1,4 +1,8 @@ +import csv +import io import json +import os +import uuid import gevent from celery.signals import task_postrun @@ -10,6 +14,7 @@ from app import create_uuid, encryption, notify_celery from app.aws import s3 from app.celery import provider_tasks from app.config import Config, QueueNames +from app.dao import notifications_dao from app.dao.inbound_sms_dao import dao_get_inbound_sms_by_id from app.dao.jobs_dao import dao_get_job_by_id, dao_update_job from app.dao.notifications_dao import ( @@ -19,7 +24,7 @@ from app.dao.notifications_dao import ( from app.dao.service_email_reply_to_dao import dao_get_reply_to_by_id from app.dao.service_inbound_api_dao import get_service_inbound_api_for_service from app.dao.service_sms_sender_dao import dao_get_service_sms_senders_by_id -from app.dao.services_dao import dao_fetch_service_by_id +from app.dao.services_dao import dao_fetch_all_services, dao_fetch_service_by_id from app.dao.templates_dao import dao_get_template_by_id from app.enums import JobStatus, KeyType, NotificationType from app.errors import TotalRequestsError @@ -32,122 +37,7 @@ from app.serialised_models import SerialisedService, SerialisedTemplate from app.service.utils import service_allowed_to_send_to from app.utils import DATETIME_FORMAT, hilite, utc_now from notifications_utils.recipients import RecipientCSV - -# @notify_celery.task(name="process-job") -# def generate_notifications_report(service_id, report_id, data, request_method): -# current_app.logger.debug(hilite("ENTER generate_notifications_report")) -# page = data["page"] if "page" in data else 1 -# page_size = ( -# data["page_size"] -# if "page_size" in data -# else current_app.config.get("PAGE_SIZE") -# ) -# # HARD CODE TO 100 for now. 1000 or 10000 causes reports to time out before they complete (if big) -# # Tests are relying on the value in config (20), whereas the UI seems to pass 10000 -# if page_size > 100: -# page_size = 100 -# limit_days = data.get("limit_days") -# include_jobs = data.get("include_jobs", True) -# include_from_test_key = data.get("include_from_test_key", False) -# include_one_off = data.get("include_one_off", True) - -# # count_pages is not being used for whether to count the number of pages, but instead as a flag -# # for whether to show pagination links -# count_pages = data.get("count_pages", True) - -# current_app.logger.debug( -# f"get pagination with {service_id} service_id filters {data} \ -# limit_days {limit_days} include_jobs {include_jobs} include_one_off {include_one_off}" -# ) -# start_time = time.time() -# current_app.logger.debug(f"Start report generation with page.size {page_size}") -# pagination = notifications_dao.get_notifications_for_service( -# service_id, -# filter_dict=data, -# page=page, -# page_size=page_size, -# count_pages=False, -# limit_days=limit_days, -# include_jobs=include_jobs, -# include_from_test_key=include_from_test_key, -# include_one_off=include_one_off, -# ) -# current_app.logger.debug(f"Query complete at {int(time.time()-start_time)*1000}") - -# for notification in pagination.items: -# if notification.job_id is not None: -# current_app.logger.debug( -# f"Processing job_id {notification.job_id} at {int(time.time()-start_time)*1000}" -# ) -# 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 - -# else: -# notification.to = "" -# notification.normalised_to = "" - -# kwargs = request.args.to_dict() -# kwargs["service_id"] = service_id - -# if data.get("format_for_csv"): -# notifications = [ -# notification.serialize_for_csv() for notification in pagination.items -# ] -# else: -# notifications = notification_with_template_schema.dump( -# pagination.items, many=True -# ) -# current_app.logger.debug(f"number of notifications are {len(notifications)}") - -# # We try and get the next page of results to work out if we need provide a pagination link to the next page -# # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous -# # call to be True which will enable us to use Flask-Sqlalchemy to tell if there is a next page of results but -# # this way is much more performant for services with many results (unlike Flask SqlAlchemy, this approach -# # doesn't do an additional query to count all the results of which there could be millions but instead only -# # asks for a single extra page of results). -# next_page_of_pagination = notifications_dao.get_notifications_for_service( -# service_id, -# filter_dict=data, -# page=page + 1, -# page_size=page_size, -# count_pages=False, -# limit_days=limit_days, -# include_jobs=include_jobs, -# include_from_test_key=include_from_test_key, -# include_one_off=include_one_off, -# error_out=False, # False so that if there are no results, it doesn't end in aborting with a 404 -# ) - -# x = ( -# jsonify( -# notifications=notifications, -# page_size=page_size, -# links=( -# get_prev_next_pagination_links( -# page, -# len(next_page_of_pagination.items), -# ".get_all_notifications_for_service", -# **kwargs, -# ) -# if count_pages -# else {} -# ), -# ), -# 200, -# ) -# current_app.logger.debug(x) +from notifications_utils.s3 import s3upload @notify_celery.task(name="process-job") @@ -662,3 +552,124 @@ def process_incomplete_job(job_id): process_row(row, template, job, job.service, sender_id=sender_id) job_complete(job, resumed=True) + + +def _generate_notifications_report(service_id, report_id, limit_days): + current_app.logger.debug(hilite("ENTER _generate_notifications_report()")) + page = 1 + page_size = 20000 + include_jobs = True + include_from_test_key = False + include_one_off = True + + data = { + "limit_days": limit_days, + "include_jobs": True, + "include_from_test_key": False, + "include_one_off": True, + } + pagination = notifications_dao.get_notifications_for_service( + service_id, + filter_dict=data, + page=page, + page_size=page_size, + count_pages=False, + limit_days=limit_days, + include_jobs=include_jobs, + include_from_test_key=include_from_test_key, + include_one_off=include_one_off, + ) + count = 1 + for notification in pagination.items: + if notification.job_id is not None: + current_app.logger.debug( + f"Processing job_id {notification.job_id} which is row {count}" + ) + count = count + 1 + notification.personalisation = s3.get_personalisation_from_s3( + notification.service_id, + notification.job_id, + notification.job_row_number, + ) + + recipient = s3.get_phone_number_from_s3( + notification.service_id, + notification.job_id, + notification.job_row_number, + ) + + notification.to = recipient + notification.normalised_to = recipient + + else: + notification.to = "" + notification.normalised_to = "" + + notifications = [ + notification.serialize_for_csv() for notification in pagination.items + ] + current_app.logger.debug(hilite(f"NUMBER OF NOTFICIATIONS IS {len(notifications)}")) + + # We try and get the next page of results to work out if we need provide a pagination link to the next page + # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous + # call to be True which will enable us to use Flask-Sqlalchemy to tell if there is a next page of results but + # this way is much more performant for services with many results (unlike Flask SqlAlchemy, this approach + # doesn't do an additional query to count all the results of which there could be millions but instead only + # asks for a single extra page of results). + + csv_bytes = io.BytesIO() + text_wrapper = io.TextIOWrapper(csv_bytes, encoding="utf-8", newline="") + writer = csv.writer(text_wrapper) + writer.writerows(notifications) + text_wrapper.flush() + csv_bytes.seek(0) + + bucket_name, file_location, access_key, secret_key, region = get_csv_location( + service_id, report_id + ) + if bucket_name == "": + exp_bucket = current_app.config["CSV_UPLOAD_BUCKET"]["bucket"] + exp_region = current_app.config["CSV_UPLOAD_BUCKET"]["region"] + tier = os.getenv("NOTIFY_ENVIRONMENT") + raise Exception( + f"NO BUCKET NAME SHOULD BE: {exp_bucket} WITH REGION {exp_region} TIER {tier}" + ) + + current_app.logger.debug(hilite(f"UPLOADING THIS {file_location}")) + s3upload( + filedata=csv_bytes, + region=region, + bucket_name=bucket_name, + file_location=file_location, + access_key=access_key, + secret_key=secret_key, + ) + current_app.logger.debug(hilite("FINITO")) + + +@notify_celery.task(name="generate-notifications-reports") +def generate_notification_reports_task(): + current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) + services = dao_fetch_all_services(only_active=True) + for service in services: + + current_app.logger.debug(hilite("INVOKE APPLY_ASYNC")) + limit_days = [1, 3, 5, 7] + for limit_day in limit_days: + + report_id = f"{str(uuid.uuid4())}-{limit_day}-day" + _generate_notifications_report(service.id, report_id, limit_day) + current_app.logger.info("Notifications report generation complete") + + +NEW_FILE_LOCATION_STRUCTURE = "{}-service-notify/{}.csv" + + +def get_csv_location(service_id, upload_id): + return ( + current_app.config["CSV_UPLOAD_BUCKET"]["bucket"], + NEW_FILE_LOCATION_STRUCTURE.format(service_id, upload_id), + current_app.config["CSV_UPLOAD_BUCKET"]["access_key_id"], + current_app.config["CSV_UPLOAD_BUCKET"]["secret_access_key"], + current_app.config["CSV_UPLOAD_BUCKET"]["region"], + ) diff --git a/app/commands.py b/app/commands.py index e4fdc7cb5..020b0610d 100644 --- a/app/commands.py +++ b/app/commands.py @@ -18,7 +18,10 @@ from sqlalchemy.orm.exc import NoResultFound from app import db, redis_store from app.aws import s3 from app.celery.nightly_tasks import cleanup_unfinished_jobs -from app.celery.tasks import process_row +from app.celery.tasks import ( + generate_notification_reports_task, + process_row, +) from app.dao.annual_billing_dao import ( dao_create_or_update_annual_billing_for_year, set_default_free_allowance_for_service, @@ -810,6 +813,11 @@ def update_templates(): _clear_templates_from_cache() +@notify_command(name="generate-notification-reports") +def generate_notification_reports(): + generate_notification_reports_task() + + def _clear_templates_from_cache(): # When we update-templates in the db, we need to make sure to delete them # from redis, otherwise the old versions will stick around forever. diff --git a/app/config.py b/app/config.py index b203fe4f9..16267c63e 100644 --- a/app/config.py +++ b/app/config.py @@ -287,6 +287,11 @@ class Config(object): "schedule": crontab(minute="*/30"), "options": {"queue": QueueNames.PERIODIC}, }, + "generate-notifications-reports": { + "task": "generate-notifications-reports", + "schedule": crontab(hour=1, minute=0), + "options": {"queue": QueueNames.PERIODIC}, + }, "regenerate-job-cache-on-startup": { "task": "regenerate-job-cache", "schedule": crontab( diff --git a/app/service/rest.py b/app/service/rest.py index c4d94e0d1..aa1d89be5 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -1,10 +1,6 @@ -import csv -import io import itertools import logging -import os import time -import uuid from datetime import datetime, timedelta from zoneinfo import ZoneInfo @@ -15,7 +11,7 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm.exc import NoResultFound from werkzeug.datastructures import MultiDict -from app import db, notify_celery +from app import db from app.aws.s3 import get_personalisation_from_s3, get_phone_number_from_s3 from app.config import QueueNames from app.dao import fact_notification_status_dao, notifications_dao @@ -124,10 +120,8 @@ from app.user.users_schema import post_set_permissions_schema from app.utils import ( check_suspicious_id, get_prev_next_pagination_links, - hilite, utc_now, ) -from notifications_utils.s3 import s3upload celery_logger = logging.getLogger(__name__) @@ -135,18 +129,6 @@ service_blueprint = Blueprint("service", __name__) register_errors(service_blueprint) -NEW_FILE_LOCATION_STRUCTURE = "{}-service-notify/{}.csv" - - -def get_csv_location(service_id, upload_id): - return ( - current_app.config["CSV_UPLOAD_BUCKET"]["bucket"], - NEW_FILE_LOCATION_STRUCTURE.format(service_id, upload_id), - current_app.config["CSV_UPLOAD_BUCKET"]["access_key_id"], - current_app.config["CSV_UPLOAD_BUCKET"]["secret_access_key"], - current_app.config["CSV_UPLOAD_BUCKET"]["region"], - ) - @service_blueprint.errorhandler(IntegrityError) def handle_integrity_error(exc): @@ -536,131 +518,7 @@ def get_service_history(service_id): return jsonify(data=data) -@notify_celery.task(name="generate-notifications-report") -def generate_notifications_report( - service_id, report_id, data, request_method, request_args -): - current_app.logger.debug(hilite("ENTER generate_notifications_report")) - page = 1 - page_size = 20000 - limit_days = data.get("limit_days") - include_jobs = data.get("include_jobs", True) - include_from_test_key = data.get("include_from_test_key", False) - include_one_off = data.get("include_one_off", True) - - current_app.logger.debug(hilite("GET ALL PARAMS")) - current_app.logger.debug( - f"get pagination with {service_id} service_id filters {data} \ - limit_days {limit_days} include_jobs {include_jobs} include_one_off {include_one_off}" - ) - start_time = time.time() - current_app.logger.debug(f"Start report generation with page.size {page_size}") - pagination = notifications_dao.get_notifications_for_service( - service_id, - filter_dict=data, - page=page, - page_size=page_size, - count_pages=False, - limit_days=limit_days, - include_jobs=include_jobs, - include_from_test_key=include_from_test_key, - include_one_off=include_one_off, - ) - current_app.logger.debug(f"Query complete at {int(time.time()-start_time)*1000}") - current_app.logger.debug(f"HOW MANY ITEMS IN PAGINATION? {len(pagination.items)}") - count = 1 - for notification in pagination.items: - if notification.job_id is not None: - current_app.logger.debug( - f"Processing job_id {notification.job_id} which is row {count}" - ) - count = count + 1 - 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 - - else: - notification.to = "" - notification.normalised_to = "" - - kwargs = request_args - kwargs["service_id"] = service_id - - notifications = [ - notification.serialize_for_csv() for notification in pagination.items - ] - current_app.logger.debug(hilite(f"NUMBER OF NOTFICIATIONS IS {len(notifications)}")) - - # We try and get the next page of results to work out if we need provide a pagination link to the next page - # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous - # call to be True which will enable us to use Flask-Sqlalchemy to tell if there is a next page of results but - # this way is much more performant for services with many results (unlike Flask SqlAlchemy, this approach - # doesn't do an additional query to count all the results of which there could be millions but instead only - # asks for a single extra page of results). - - csv_bytes = io.BytesIO() - text_wrapper = io.TextIOWrapper(csv_bytes, encoding="utf-8", newline="") - writer = csv.writer(text_wrapper) - writer.writerows(notifications) - text_wrapper.flush() - csv_bytes.seek(0) - - bucket_name, file_location, access_key, secret_key, region = get_csv_location( - service_id, report_id - ) - if bucket_name == "": - exp_bucket = current_app.config["CSV_UPLOAD_BUCKET"]["bucket"] - exp_region = current_app.config["CSV_UPLOAD_BUCKET"]["region"] - tier = os.getenv("NOTIFY_ENVIRONMENT") - raise Exception( - f"NO BUCKET NAME SHOULD BE: {exp_bucket} WITH REGION {exp_region} TIER {tier}" - ) - - current_app.logger.debug(f"UPLOADING THIS {csv_bytes}") - s3upload( - filedata=csv_bytes, - region=region, - bucket_name=bucket_name, - file_location=file_location, - access_key=access_key, - secret_key=secret_key, - ) - current_app.logger.debug(hilite("FINITO")) - - -@service_blueprint.route("//notifications-report", methods=["GET", "POST"]) -def get_notifications_report_for_service(service_id): - current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) - check_suspicious_id(service_id) - report_id = str(uuid.uuid4()) - request_method = request.method - if request_method == "GET": - data = notifications_filter_schema.load(request.args) - else: - data = notifications_filter_schema.load( - MultiDict(request.get_json(silent=True)) - ) - request_args = request.args.to_dict() - current_app.logger.debug(hilite("INVOKE APPLY_ASYNC")) - generate_notifications_report.apply_async( - args=[service_id, report_id, data, request_method, request_args], - queue=QueueNames.NOTIFY, - ) - return jsonify({"report_id": report_id}), 200 - - -@service_blueprint.route("//notifications", methods=["GET", "POST"]) +@service_blueprint.route("//notificationsx", methods=["GET", "POST"]) def get_all_notifications_for_service(service_id): check_suspicious_id(service_id) current_app.logger.debug("enter get_all_notifications_for_service") From 0bcfb886bc3863373f7973e5214604fee2b1aba4 Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Tue, 5 Aug 2025 13:53:07 -0700 Subject: [PATCH 7/8] fix typo --- app/service/rest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/service/rest.py b/app/service/rest.py index aa1d89be5..f50e7afcc 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -518,7 +518,7 @@ def get_service_history(service_id): return jsonify(data=data) -@service_blueprint.route("//notificationsx", methods=["GET", "POST"]) +@service_blueprint.route("//notifications", methods=["GET", "POST"]) def get_all_notifications_for_service(service_id): check_suspicious_id(service_id) current_app.logger.debug("enter get_all_notifications_for_service") From 9f8470230158ee6258db69cbb2a04111769840a9 Mon Sep 17 00:00:00 2001 From: Kenneth Kehl <@kkehl@flexion.us> Date: Wed, 6 Aug 2025 07:19:18 -0700 Subject: [PATCH 8/8] cleanup --- app/aws/s3.py | 38 ++++++++++++++++++++++++++++++++++++++ app/celery/tasks.py | 19 +++++++++++-------- 2 files changed, 49 insertions(+), 8 deletions(-) diff --git a/app/aws/s3.py b/app/aws/s3.py index 9c397fc1a..8ca10676e 100644 --- a/app/aws/s3.py +++ b/app/aws/s3.py @@ -123,10 +123,48 @@ def list_s3_objects(): ) +def get_notification_reports(service_id): + + bucket_name = _get_bucket_name() + s3_client = get_s3_client() + # Our reports only support 7 days, but pull 8 days to avoid + # any edge cases + time_limit = aware_utcnow() - datetime.timedelta(days=8) + reports = [] + try: + response = s3_client.list_objects_v2(Bucket=bucket_name) + while True: + for obj in response.get("Contents", []): + if obj["LastModified"] >= time_limit: + if service_id in obj["Key"] and "report" in obj["Key"]: + reports.append(obj) + if "NextContinuationToken" in response: + response = s3_client.list_objects_v2( + Bucket=bucket_name, + ContinuationToken=response["NextContinuationToken"], + ) + else: + break + except Exception as e: + current_app.logger.exception( + f"An error occurred while regenerating cache #notify-debug-admin-1200: {str(e)}", + ) + return reports + + def get_bucket_name(): return current_app.config["CSV_UPLOAD_BUCKET"]["bucket"] +def delete_s3_object(key): + + try: + remove_csv_object(key) + current_app.logger.debug(f"#delete-s3-object Deleted: {key}") + except botocore.exceptions.ClientError: + current_app.logger.exception(f"Couldn't delete {key}") + + def cleanup_old_s3_objects(): bucket_name = get_bucket_name() diff --git a/app/celery/tasks.py b/app/celery/tasks.py index b665f2d09..f11a13e54 100644 --- a/app/celery/tasks.py +++ b/app/celery/tasks.py @@ -2,7 +2,6 @@ import csv import io import json import os -import uuid import gevent from celery.signals import task_postrun @@ -555,7 +554,8 @@ def process_incomplete_job(job_id): def _generate_notifications_report(service_id, report_id, limit_days): - current_app.logger.debug(hilite("ENTER _generate_notifications_report()")) + + # Hard code these values for now page = 1 page_size = 20000 include_jobs = True @@ -608,7 +608,9 @@ def _generate_notifications_report(service_id, report_id, limit_days): notifications = [ notification.serialize_for_csv() for notification in pagination.items ] - current_app.logger.debug(hilite(f"NUMBER OF NOTFICIATIONS IS {len(notifications)}")) + current_app.logger.debug( + hilite(f"Number of notifications in report: {len(notifications)}") + ) # We try and get the next page of results to work out if we need provide a pagination link to the next page # in our response if it exists. Note, this could be done instead by changing `count_pages` in the previous @@ -632,10 +634,12 @@ def _generate_notifications_report(service_id, report_id, limit_days): exp_region = current_app.config["CSV_UPLOAD_BUCKET"]["region"] tier = os.getenv("NOTIFY_ENVIRONMENT") raise Exception( - f"NO BUCKET NAME SHOULD BE: {exp_bucket} WITH REGION {exp_region} TIER {tier}" + f"No bucket name should be: {exp_bucket} with region {exp_region} and tier {tier}" ) - current_app.logger.debug(hilite(f"UPLOADING THIS {file_location}")) + # Delete yesterday's version of this report + s3.delete_s3_object(file_location) + s3upload( filedata=csv_bytes, region=region, @@ -644,12 +648,11 @@ def _generate_notifications_report(service_id, report_id, limit_days): access_key=access_key, secret_key=secret_key, ) - current_app.logger.debug(hilite("FINITO")) + current_app.logger.info(f"generate-notifications-report uploaded {file_location}") @notify_celery.task(name="generate-notifications-reports") def generate_notification_reports_task(): - current_app.logger.debug(hilite("ENTER GET ALL NOTIFICATIONS FOR SERVICE@")) services = dao_fetch_all_services(only_active=True) for service in services: @@ -657,7 +660,7 @@ def generate_notification_reports_task(): limit_days = [1, 3, 5, 7] for limit_day in limit_days: - report_id = f"{str(uuid.uuid4())}-{limit_day}-day" + report_id = f"{limit_day}-day-report" _generate_notifications_report(service.id, report_id, limit_day) current_app.logger.info("Notifications report generation complete")