mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-22 00:11:16 -05:00
ad274ee887
This is a similar PR to https://github.com/alphagov/notifications-api/pull/2284. When using flask-sqlalchemy to get a `Pagination` object, by default it will run two queries 1. Get the page of results that you are asking for 2. If the number of results is equal to the page size, then it will issue a second query that will count the total number of results Getting the total number of results is only useful if - you need to show how many results there are - you need to know if there is a next page of results (flask-sqlalchemy uses the total to work out how many pages there are altogether, which may not be the most efficient way of working out if there is a next page or not but that is what it currently does). Looking at the `get_notifications` route, it does not use `paginated_notifications.total` or `paginated_notifications.has_next` and therefore we have no use for the second query to get the total number of results. We can stop this additional query by setting `count_pages=False` which will hopefully give us some performance improvements, in particular for services which send a lot of notifications. Flask sqlalchemy references:818c947b66/src/flask_sqlalchemy/__init__.py (L478)818c947b66/src/flask_sqlalchemy/__init__.py (L399)Note, I have checked the other uses of `get_notifications_for_service` and the other cases are currently using the total or next page so this approach is not something we can take with them.
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
from io import BytesIO
|
|
|
|
from flask import current_app, jsonify, request, send_file, url_for
|
|
|
|
from app import api_user, authenticated_service
|
|
from app.dao import notifications_dao
|
|
from app.letters.utils import get_letter_pdf_and_metadata
|
|
from app.models import (
|
|
LETTER_TYPE,
|
|
NOTIFICATION_PENDING_VIRUS_CHECK,
|
|
NOTIFICATION_TECHNICAL_FAILURE,
|
|
NOTIFICATION_VIRUS_SCAN_FAILED,
|
|
)
|
|
from app.schema_validation import validate
|
|
from app.v2.errors import BadRequestError, PDFNotReadyError
|
|
from app.v2.notifications import v2_notification_blueprint
|
|
from app.v2.notifications.notification_schemas import (
|
|
get_notifications_request,
|
|
notification_by_id,
|
|
)
|
|
|
|
|
|
@v2_notification_blueprint.route("/<notification_id>", methods=['GET'])
|
|
def get_notification_by_id(notification_id):
|
|
_data = {"notification_id": notification_id}
|
|
validate(_data, notification_by_id)
|
|
notification = notifications_dao.get_notification_with_personalisation(
|
|
authenticated_service.id, notification_id, key_type=None
|
|
)
|
|
return jsonify(notification.serialize()), 200
|
|
|
|
|
|
@v2_notification_blueprint.route('/<notification_id>/pdf', methods=['GET'])
|
|
def get_pdf_for_notification(notification_id):
|
|
_data = {"notification_id": notification_id}
|
|
validate(_data, notification_by_id)
|
|
notification = notifications_dao.get_notification_by_id(
|
|
notification_id, authenticated_service.id, _raise=True
|
|
)
|
|
|
|
if notification.notification_type != LETTER_TYPE:
|
|
raise BadRequestError(message="Notification is not a letter")
|
|
|
|
if notification.status == NOTIFICATION_VIRUS_SCAN_FAILED:
|
|
raise BadRequestError(message='File did not pass the virus scan')
|
|
|
|
if notification.status == NOTIFICATION_TECHNICAL_FAILURE:
|
|
raise BadRequestError(message='PDF not available for letters in status {}'.format(notification.status))
|
|
|
|
if notification.status == NOTIFICATION_PENDING_VIRUS_CHECK:
|
|
raise PDFNotReadyError()
|
|
|
|
try:
|
|
pdf_data, metadata = get_letter_pdf_and_metadata(notification)
|
|
except Exception:
|
|
raise PDFNotReadyError()
|
|
|
|
return send_file(filename_or_fp=BytesIO(pdf_data), mimetype='application/pdf')
|
|
|
|
|
|
@v2_notification_blueprint.route("", methods=['GET'])
|
|
def get_notifications():
|
|
_data = request.args.to_dict(flat=False)
|
|
|
|
# flat=False makes everything a list, but we only ever allow one value for "older_than"
|
|
if 'older_than' in _data:
|
|
_data['older_than'] = _data['older_than'][0]
|
|
|
|
# and client reference
|
|
if 'reference' in _data:
|
|
_data['reference'] = _data['reference'][0]
|
|
|
|
if 'include_jobs' in _data:
|
|
_data['include_jobs'] = _data['include_jobs'][0]
|
|
|
|
data = validate(_data, get_notifications_request)
|
|
|
|
paginated_notifications = notifications_dao.get_notifications_for_service(
|
|
str(authenticated_service.id),
|
|
filter_dict=data,
|
|
key_type=api_user.key_type,
|
|
personalisation=True,
|
|
older_than=data.get('older_than'),
|
|
client_reference=data.get('reference'),
|
|
page_size=current_app.config.get('API_PAGE_SIZE'),
|
|
include_jobs=data.get('include_jobs'),
|
|
count_pages=False
|
|
)
|
|
|
|
def _build_links(notifications):
|
|
_links = {
|
|
'current': url_for(".get_notifications", _external=True, **data),
|
|
}
|
|
|
|
if len(notifications):
|
|
next_query_params = dict(data, older_than=notifications[-1].id)
|
|
_links['next'] = url_for(".get_notifications", _external=True, **next_query_params)
|
|
|
|
return _links
|
|
|
|
return jsonify(
|
|
notifications=[notification.serialize() for notification in paginated_notifications.items],
|
|
links=_build_links(paginated_notifications.items)
|
|
), 200
|