Skip to content

wip report request process #4898

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
spatel033 wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

wip report request process #4898

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
250 changes: 142 additions & 108 deletions app/report_requests/process_notifications_report.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import csv
from io import StringIO
from datetime import datetime, timedelta
from io import BytesIO
from typing import Any
from uuid import UUID

Expand All @@ -14,137 +14,173 @@

from app import db
from app.constants import NOTIFICATION_REPORT_REQUEST_MAPPING
from app.dao.notifications_dao import get_notifications_for_service
from app.dao.report_requests_dao import dao_get_report_request_by_id
from app.dao.service_data_retention_dao import fetch_service_data_retention_by_notification_type
from app.models import (
Notification,
)
from app.utils import (
midnight_n_days_ago,
)


class ReportRequestProcessor:
def __init__(self, service_id: UUID, report_request_id: UUID):
self.service_id = service_id
self.report_request_id = report_request_id
self.report_request = dao_get_report_request_by_id(service_id, report_request_id)

self.notification_type = self.report_request.parameter["notification_type"]
self.notification_status = self.report_request.parameter["notification_status"]
self.page_size = current_app.config.get("REPORT_REQUEST_NOTIFICATIONS_CSV_BATCH_SIZE")
self.s3_bucket = current_app.config["S3_BUCKET_REPORT_REQUESTS_DOWNLOAD"]
self.filename = f"notifications_report/{report_request_id}.csv"

self.upload_id: str | None = None
self.parts: list[dict[str, Any]] = []
self.part_number = 1
self.csv_buffer = StringIO()
self.csv_writer = csv.writer(self.csv_buffer)
self.csv_buffer = BytesIO()

def process(self) -> None:
self._initialize_csv()
self._start_multipart_upload()

try:
self._fetch_and_upload_notifications()
self._stream_query_to_s3()
self._finalize_upload()
except Exception as e:
current_app.logger.exception("Error occurred while processing the report: %s", e)
self._abort_upload()
raise e

def _initialize_csv(self) -> None:
headers = [
"Recipient",
"Reference",
"Template",
"Type",
"Sent by",
"Sent by email",
"Job",
"Status",
"Time",
"API key name",
]
self.csv_writer.writerow(headers)
def write(self, data: bytes) -> None:
self.csv_buffer.write(data)
if self.csv_buffer.tell() >= S3_MULTIPART_UPLOAD_MIN_PART_SIZE:
self._upload_csv_part()

def _start_multipart_upload(self) -> None:
response = s3_multipart_upload_create(self.s3_bucket, self.filename)
self.upload_id = response["UploadId"]

def _fetch_and_upload_notifications(self) -> None:
def _stream_query_to_s3(self) -> None:
service_retention = fetch_service_data_retention_by_notification_type(self.service_id, self.notification_type)
limit_days = service_retention.days_of_retention if service_retention else 7
older_than = None
is_notification = True
while is_notification:
serialized_notifications = self._fetch_serialized_notifications(limit_days, older_than)

is_notification = len(serialized_notifications) != 0

csv_data = self._convert_notifications_to_csv(serialized_notifications)
self.csv_writer.writerows(csv_data)
self._upload_csv_part_if_needed()
older_than = serialized_notifications[-1]["id"] if is_notification else None
# Upload any remaining data
self._upload_remaining_data()

def _fetch_serialized_notifications(self, limit_days: int, older_than: str | None) -> list[dict[str, Any]]:
statuses = NOTIFICATION_REPORT_REQUEST_MAPPING[self.notification_status]

notifications = get_notifications_for_service( # type: ignore[call-arg]
service_id=self.service_id,
filter_dict={
"template_type": self.notification_type,
"status": statuses,
},
page_size=self.page_size,
count_pages=False,
limit_days=limit_days,
include_jobs=True,
with_personalisation=False,
include_from_test_key=False,
error_out=False,
include_one_off=True,
older_than=older_than,
session=db.session_bulk,
retry_attempts=2,
)

serialized_notifications = [notification.serialize_for_csv() for notification in notifications]
return serialized_notifications

def _convert_notifications_to_csv(self, serialized_notifications: list[dict[str, Any]]) -> list[tuple]:
values = []
for notification in serialized_notifications:
values.append(
(
# the recipient for precompiled letters is the full address block
notification["recipient"].splitlines()[0].lstrip().rstrip(" ,"),
notification["client_reference"],
notification["template_name"],
notification["template_type"],
notification["created_by_name"] or "",
notification["created_by_email_address"] or "",
notification["job_name"] or "",
notification["status"],
notification["created_at"],
notification["api_key_name"] or "",
)
requested_statuses = NOTIFICATION_REPORT_REQUEST_MAPPING[self.notification_status]
filtered_statuses = Notification.substitute_status(requested_statuses)

sa_connection = db.session.connection()
raw_conn = sa_connection.connection
cursor = raw_conn.cursor()

try:
start_time = midnight_n_days_ago(limit_days)
end_time = datetime.utcnow()
current_app.logger.info(
"Generating report for service %s from %s to %s now %s",
self.service_id,
start_time,
end_time,
datetime.utcnow(),
)
return values
chunk_interval = timedelta(hours=1)
current_chunk_start = start_time

is_first_chunk = True

while current_chunk_start < end_time:
current_chunk_end = min(current_chunk_start + chunk_interval, end_time)

csv_format = "WITH CSV HEADER" if is_first_chunk else "WITH CSV"
is_first_chunk = False

sql = f"""
COPY (
SELECT
n.to AS "Recipient",
n.client_reference AS "Reference",
t.name AS "Template",
n.notification_type AS "Type",
u.name AS "Sent by",
u.email_address AS "Sent by email",
j.original_file_name AS "Job",
CASE
WHEN n.notification_type = 'email' THEN
CASE n.notification_status
WHEN 'failed' THEN 'Failed'
WHEN 'technical-failure' THEN 'Technical failure'
WHEN 'temporary-failure' THEN 'Inbox not accepting messages right now'
WHEN 'permanent-failure' THEN 'Email address doesn’t exist'
WHEN 'delivered' THEN 'Delivered'
WHEN 'sending' THEN 'Sending'
WHEN 'created' THEN 'Sending'
WHEN 'sent' THEN 'Delivered'
ELSE n.notification_status
END
WHEN n.notification_type = 'sms' THEN
CASE n.notification_status
WHEN 'failed' THEN 'Failed'
WHEN 'technical-failure' THEN 'Technical failure'
WHEN 'temporary-failure' THEN 'Phone not accepting messages right now'
WHEN 'permanent-failure' THEN 'Phone number doesn’t exist'
WHEN 'delivered' THEN 'Delivered'
WHEN 'sending' THEN 'Sending'
WHEN 'created' THEN 'Sending'
WHEN 'sent' THEN 'Sent internationally'
ELSE n.notification_status
END
WHEN n.notification_type = 'letter' THEN
CASE n.notification_status
WHEN 'technical-failure' THEN 'Technical failure'
WHEN 'permanent-failure' THEN 'Permanent failure'
WHEN 'sending' THEN 'Accepted'
WHEN 'created' THEN 'Accepted'
WHEN 'delivered' THEN 'Received'
WHEN 'returned-letter' THEN 'Returned'
ELSE n.notification_status
END
ELSE n.notification_status
END AS "Status",
n.created_at AS "Time",
a.name AS "API key name"
FROM notifications n
LEFT JOIN templates_history t ON n.template_id = t.id AND n.template_version = t.version
LEFT JOIN users u ON n.created_by_id = u.id
LEFT JOIN jobs j ON n.job_id = j.id
LEFT JOIN api_keys a ON n.api_key_id = a.id
WHERE n.service_id = %(service_id)s
AND n.notification_type = %(notification_type)s
AND n.notification_status IN %(statuses)s
AND n.created_at >= %(chunk_start)s
AND n.created_at < %(chunk_end)s
ORDER BY n.created_at ASC
) TO STDOUT {csv_format};
"""

# Note: psycopg2 requires lists to be converted to tuples for the IN clause
bound_sql = cursor.mogrify(
sql,
{
"service_id": str(self.service_id),
"notification_type": self.notification_type,
"statuses": tuple(filtered_statuses),
"chunk_start": current_chunk_start,
"chunk_end": current_chunk_end,
},
)

def _upload_csv_part_if_needed(self) -> None:
data_bytes = self.csv_buffer.getvalue().encode("utf-8")
if len(data_bytes) >= S3_MULTIPART_UPLOAD_MIN_PART_SIZE:
self._upload_part(data_bytes)
current_app.logger.info("Processing chunk from %s to %s", current_chunk_start, current_chunk_end)

# Reset the buffer for the next part
# truncate(0) does not reset the cursor so seek(0) is needed to reset the cursor
self.csv_buffer.seek(0)
self.csv_buffer.truncate(0)
self.csv_writer = csv.writer(self.csv_buffer)
cursor.copy_expert(bound_sql, self)
current_chunk_start = current_chunk_end

def _upload_remaining_data(self) -> None:
data_bytes = self.csv_buffer.getvalue().encode("utf-8")
if len(data_bytes) > 0:
self._upload_part(data_bytes)
if self.csv_buffer.tell() > 0:
self._upload_csv_part()

finally:
cursor.close()

def _upload_csv_part(self) -> None:
data_bytes = self.csv_buffer.getvalue()

def _upload_part(self, data_bytes: bytes) -> None:
response = s3_multipart_upload_part(
part_number=self.part_number,
bucket_name=self.s3_bucket,
Expand All @@ -168,29 +204,27 @@ def _upload_part(self, data_bytes: bytes) -> None:
self.parts.append({"PartNumber": self.part_number, "ETag": response["ETag"]})
self.part_number += 1

self.csv_buffer.seek(0)
self.csv_buffer.truncate(0)

def _finalize_upload(self) -> None:
s3_multipart_upload_complete(
bucket_name=self.s3_bucket,
filename=self.filename,
upload_id=self.upload_id,
parts=self.parts,
)
extra = {
"report_request_id": self.report_request_id,
"s3_bucket": self.s3_bucket,
"s3_key": self.filename,
"part_count": len(self.parts),
}
current_app.logger.info(
"Upload complete for report request %(report_request_id)s to bucket %(s3_bucket)s "
"with filename %(s3_key)s. Total parts: %(part_count)s.",
extra,
extra=extra,
"Upload complete for report request %s to bucket %s. Total parts: %s.",
self.report_request_id,
self.s3_bucket,
len(self.parts),
)

def _abort_upload(self) -> None:
s3_multipart_upload_abort(
bucket_name=self.s3_bucket,
filename=self.filename,
upload_id=self.upload_id,
)
if self.upload_id:
s3_multipart_upload_abort(
bucket_name=self.s3_bucket,
filename=self.filename,
upload_id=self.upload_id,
)
Loading