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

merge from main

Browse Source
This commit is contained in:
Kenneth Kehl
2024-10-29 12:20:46 -07:00
parent f8f4e46f48 f782f1ec64
commit e5117bb969
31 changed files with 1823 additions and 971 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
app/aws/s3.py
View File

@@ -476,23 +476,7 @@ def get_personalisation_from_s3(service_id, job_id, job_row_number):
set_job_cache(job_cache, f"{job_id}_personalisation", extract_personalisation(job))
# If we can find the quick dictionary, use it
if job_cache.get(f"{job_id}_personalisation") is not None:
personalisation_to_return = job_cache.get(f"{job_id}_personalisation")[0].get(
job_row_number
)
if personalisation_to_return:
return personalisation_to_return
else:
current_app.logger.warning(
f"Was unable to retrieve personalisation from lookup dictionary for job {job_id}"
)
return {}
else:
current_app.logger.error(
f"Was unable to construct lookup dictionary for job {job_id}"
)
return {}
return job_cache.get(f"{job_id}_personalisation")[0].get(job_row_number)
def get_job_metadata_from_s3(service_id, job_id):

79
app/commands.py
View File

@@ -24,12 +24,6 @@ from app.dao.annual_billing_dao import (
dao_create_or_update_annual_billing_for_year,
set_default_free_allowance_for_service,
)
from app.dao.fact_billing_dao import (
delete_billing_data_for_service_for_day,
fetch_billing_data_for_day,
get_service_ids_that_need_billing_populated,
update_fact_billing,
)
from app.dao.jobs_dao import dao_get_job_by_id
from app.dao.organization_dao import (
dao_add_service_to_organization,
@@ -63,7 +57,7 @@ from app.models import (
TemplateHistory,
User,
)
from app.utils import get_midnight_in_utc, utc_now
from app.utils import utc_now
from notifications_utils.recipients import RecipientCSV
from notifications_utils.template import SMSMessageTemplate
from tests.app.db import (
@@ -167,6 +161,7 @@ def purge_functional_test_data(user_email_prefix):
delete_model_user(usr)
# TODO maintainability what is the purpose of this command? Who would use it and why?
@notify_command(name="insert-inbound-numbers")
@click.option(
"-f",
@@ -175,7 +170,6 @@ def purge_functional_test_data(user_email_prefix):
help="""Full path of the file to upload, file is a contains inbound numbers, one number per line.""",
)
def insert_inbound_numbers_from_file(file_name):
# TODO maintainability what is the purpose of this command? Who would use it and why?
current_app.logger.info(f"Inserting inbound numbers from {file_name}")
with open(file_name) as file:
@@ -195,50 +189,6 @@ def setup_commands(application):
application.cli.add_command(command_group)
@notify_command(name="rebuild-ft-billing-for-day")
@click.option("-s", "--service_id", required=False, type=click.UUID)
@click.option(
"-d",
"--day",
help="The date to recalculate, as YYYY-MM-DD",
required=True,
type=click_dt(format="%Y-%m-%d"),
)
def rebuild_ft_billing_for_day(service_id, day):
# TODO maintainability what is the purpose of this command? Who would use it and why?
"""
Rebuild the data in ft_billing for the given service_id and date
"""
def rebuild_ft_data(process_day, service):
deleted_rows = delete_billing_data_for_service_for_day(process_day, service)
current_app.logger.info(
f"deleted {deleted_rows} existing billing rows for {service} on {process_day}"
)
transit_data = fetch_billing_data_for_day(
process_day=process_day, service_id=service
)
# transit_data = every row that should exist
for data in transit_data:
# upsert existing rows
update_fact_billing(data, process_day)
current_app.logger.info(
f"added/updated {len(transit_data)} billing rows for {service} on {process_day}"
)
if service_id:
# confirm the service exists
dao_fetch_service_by_id(service_id)
rebuild_ft_data(day, service_id)
else:
services = get_service_ids_that_need_billing_populated(
get_midnight_in_utc(day), get_midnight_in_utc(day + timedelta(days=1))
)
for row in services:
rebuild_ft_data(day, row.service_id)
@notify_command(name="bulk-invite-user-to-service")
@click.option(
"-f",
@@ -472,31 +422,6 @@ def associate_services_to_organizations():
current_app.logger.info("finished associating services to organizations")
@notify_command(name="populate-service-volume-intentions")
@click.option(
"-f",
"--file_name",
required=True,
help="Pipe delimited file containing service_id, SMS, email",
)
def populate_service_volume_intentions(file_name):
# [0] service_id
# [1] SMS:: volume intentions for service
# [2] Email:: volume intentions for service
# TODO maintainability what is the purpose of this command? Who would use it and why?
with open(file_name, "r") as f:
for line in itertools.islice(f, 1, None):
columns = line.split(",")
current_app.logger.info(columns)
service = dao_fetch_service_by_id(columns[0])
service.volume_sms = columns[1]
service.volume_email = columns[2]
dao_update_service(service)
current_app.logger.info("populate-service-volume-intentions complete")
@notify_command(name="populate-go-live")
@click.option(
"-f", "--file_name", required=True, help="CSV file containing live service data"

106
app/dao/notifications_dao.py
View File

@@ -1,7 +1,7 @@
from datetime import timedelta
from flask import current_app
from sqlalchemy import asc, desc, or_, select, text, union
from sqlalchemy import asc, delete, desc, func, or_, select, text, union, update
from sqlalchemy.orm import joinedload
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.sql import functions
@@ -109,11 +109,12 @@ def _update_notification_status(
def update_notification_status_by_id(
notification_id, status, sent_by=None, provider_response=None, carrier=None
):
notification = (
Notification.query.with_for_update()
stmt = (
select(Notification)
.with_for_update()
.filter(Notification.id == notification_id)
.first()
)
notification = db.session.execute(stmt).scalars().first()
if not notification:
current_app.logger.info(
@@ -156,9 +157,8 @@ def update_notification_status_by_id(
@autocommit
def update_notification_status_by_reference(reference, status):
# this is used to update emails
notification = Notification.query.filter(
Notification.reference == reference
).first()
stmt = select(Notification).filter(Notification.reference == reference)
notification = db.session.execute(stmt).scalars().first()
if not notification:
current_app.logger.error(
@@ -200,19 +200,20 @@ def get_notifications_for_job(
def dao_get_notification_count_for_job_id(*, job_id):
return Notification.query.filter_by(job_id=job_id).count()
stmt = select(func.count(Notification.id)).filter_by(job_id=job_id)
return db.session.execute(stmt).scalar()
def dao_get_notification_count_for_service(*, service_id):
notification_count = Notification.query.filter_by(service_id=service_id).count()
return notification_count
stmt = select(func.count(Notification.id)).filter_by(service_id=service_id)
return db.session.execute(stmt).scalar()
def dao_get_failed_notification_count():
failed_count = Notification.query.filter_by(
stmt = select(func.count(Notification.id)).filter_by(
status=NotificationStatus.FAILED
).count()
return failed_count
)
return db.session.execute(stmt).scalar()
def get_notification_with_personalisation(service_id, notification_id, key_type):
@@ -220,11 +221,12 @@ def get_notification_with_personalisation(service_id, notification_id, key_type)
if key_type:
filter_dict["key_type"] = key_type
return (
Notification.query.filter_by(**filter_dict)
stmt = (
select(Notification)
.filter_by(**filter_dict)
.options(joinedload(Notification.template))
.one()
)
return db.session.execute(stmt).scalars().one()
def get_notification_by_id(notification_id, service_id=None, _raise=False):
@@ -233,9 +235,13 @@ def get_notification_by_id(notification_id, service_id=None, _raise=False):
if service_id:
filters.append(Notification.service_id == service_id)
query = Notification.query.filter(*filters)
stmt = select(Notification).filter(*filters)
return query.one() if _raise else query.first()
return (
db.session.execute(stmt).scalars().one()
if _raise
else db.session.execute(stmt).scalars().first()
)
def get_notifications_for_service(
@@ -415,12 +421,13 @@ def move_notifications_to_notification_history(
deleted += delete_count_per_call
# Deleting test Notifications, test notifications are not persisted to NotificationHistory
Notification.query.filter(
stmt = delete(Notification).filter(
Notification.notification_type == notification_type,
Notification.service_id == service_id,
Notification.created_at < timestamp_to_delete_backwards_from,
Notification.key_type == KeyType.TEST,
).delete(synchronize_session=False)
)
db.session.execute(stmt)
db.session.commit()
return deleted
@@ -442,8 +449,9 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
current_statuses = [NotificationStatus.SENDING, NotificationStatus.PENDING]
new_status = NotificationStatus.TEMPORARY_FAILURE
notifications = (
Notification.query.filter(
stmt = (
select(Notification)
.filter(
Notification.created_at < cutoff_time,
Notification.status.in_(current_statuses),
Notification.notification_type.in_(
@@ -451,14 +459,15 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
),
)
.limit(limit)
.all()
)
notifications = db.session.execute(stmt).scalars().all()
Notification.query.filter(
Notification.id.in_([n.id for n in notifications]),
).update(
{"status": new_status, "updated_at": updated_at}, synchronize_session=False
stmt = (
update(Notification)
.filter(Notification.id.in_([n.id for n in notifications]))
.values({"status": new_status, "updated_at": updated_at})
)
db.session.execute(stmt)
db.session.commit()
return notifications
@@ -466,15 +475,23 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
@autocommit
def dao_update_notifications_by_reference(references, update_dict):
updated_count = Notification.query.filter(
Notification.reference.in_(references)
).update(update_dict, synchronize_session=False)
stmt = (
update(Notification)
.filter(Notification.reference.in_(references))
.values(update_dict)
)
result = db.session.execute(stmt)
updated_count = result.rowcount
updated_history_count = 0
if updated_count != len(references):
updated_history_count = NotificationHistory.query.filter(
NotificationHistory.reference.in_(references)
).update(update_dict, synchronize_session=False)
stmt = (
update(NotificationHistory)
.filter(NotificationHistory.reference.in_(references))
.values(update_dict)
)
result = db.session.execute(stmt)
updated_history_count = result.rowcount
return updated_count, updated_history_count
@@ -541,18 +558,21 @@ def dao_get_notifications_by_recipient_or_reference(
def dao_get_notification_by_reference(reference):
return Notification.query.filter(Notification.reference == reference).one()
stmt = select(Notification).filter(Notification.reference == reference)
return db.session.execute(stmt).scalars().one()
def dao_get_notification_history_by_reference(reference):
try:
# This try except is necessary because in test keys and research mode does not create notification history.
# Otherwise we could just search for the NotificationHistory object
return Notification.query.filter(Notification.reference == reference).one()
stmt = select(Notification).filter(Notification.reference == reference)
return db.session.execute(stmt).scalars().one()
except NoResultFound:
return NotificationHistory.query.filter(
stmt = select(NotificationHistory).filter(
NotificationHistory.reference == reference
).one()
)
return db.session.execute(stmt).scalars().one()
def dao_get_notifications_processing_time_stats(start_date, end_date):
@@ -590,11 +610,12 @@ def dao_get_notifications_processing_time_stats(start_date, end_date):
def dao_get_last_notification_added_for_job_id(job_id):
last_notification_added = (
Notification.query.filter(Notification.job_id == job_id)
stmt = (
select(Notification)
.filter(Notification.job_id == job_id)
.order_by(Notification.job_row_number.desc())
.first()
)
last_notification_added = db.session.execute(stmt).scalars().first()
return last_notification_added
@@ -602,11 +623,12 @@ def dao_get_last_notification_added_for_job_id(job_id):
def notifications_not_yet_sent(should_be_sending_after_seconds, notification_type):
older_than_date = utc_now() - timedelta(seconds=should_be_sending_after_seconds)
notifications = Notification.query.filter(
stmt = select(Notification).filter(
Notification.created_at <= older_than_date,
Notification.notification_type == notification_type,
Notification.status == NotificationStatus.CREATED,
).all()
)
notifications = db.session.execute(stmt).scalars().all()
return notifications

53
app/dao/organization_dao.py
View File

@@ -1,3 +1,4 @@
from sqlalchemy import delete, select, update
from sqlalchemy.sql.expression import func
from app import db
@@ -6,55 +7,57 @@ from app.models import Domain, Organization, Service, User
def dao_get_organizations():
return Organization.query.order_by(
stmt = select(Organization).order_by(
Organization.active.desc(), Organization.name.asc()
).all()
)
return db.session.execute(stmt).scalars().all()
def dao_count_organizations_with_live_services():
return (
db.session.query(Organization.id)
stmt = (
select(func.count(func.distinct(Organization.id)))
.join(Organization.services)
.filter(
Service.active.is_(True),
Service.restricted.is_(False),
Service.count_as_live.is_(True),
)
.distinct()
.count()
)
return db.session.execute(stmt).scalar() or 0
def dao_get_organization_services(organization_id):
return Organization.query.filter_by(id=organization_id).one().services
stmt = select(Organization).filter_by(id=organization_id)
return db.session.execute(stmt).scalars().one().services
def dao_get_organization_live_services(organization_id):
return Service.query.filter_by(
organization_id=organization_id, restricted=False
).all()
stmt = select(Service).filter_by(organization_id=organization_id, restricted=False)
return db.session.execute(stmt).scalars().all()
def dao_get_organization_by_id(organization_id):
return Organization.query.filter_by(id=organization_id).one()
stmt = select(Organization).filter_by(id=organization_id)
return db.session.execute(stmt).scalars().one()
def dao_get_organization_by_email_address(email_address):
email_address = email_address.lower().replace(".gsi.gov.uk", ".gov.uk")
for domain in Domain.query.order_by(func.char_length(Domain.domain).desc()).all():
stmt = select(Domain).order_by(func.char_length(Domain.domain).desc())
domains = db.session.execute(stmt).scalars().all()
for domain in domains:
if email_address.endswith(
"@{}".format(domain.domain)
) or email_address.endswith(".{}".format(domain.domain)):
return Organization.query.filter_by(id=domain.organization_id).one()
stmt = select(Organization).filter_by(id=domain.organization_id)
return db.session.execute(stmt).scalars().one()
return None
def dao_get_organization_by_service_id(service_id):
return (
Organization.query.join(Organization.services).filter_by(id=service_id).first()
)
stmt = select(Organization).join(Organization.services).filter_by(id=service_id)
return db.session.execute(stmt).scalars().first()
@autocommit
@@ -65,10 +68,14 @@ def dao_create_organization(organization):
@autocommit
def dao_update_organization(organization_id, **kwargs):
domains = kwargs.pop("domains", None)
num_updated = Organization.query.filter_by(id=organization_id).update(kwargs)
stmt = (
update(Organization).where(Organization.id == organization_id).values(**kwargs)
)
num_updated = db.session.execute(stmt).rowcount
if isinstance(domains, list):
Domain.query.filter_by(organization_id=organization_id).delete()
stmt = delete(Domain).filter_by(organization_id=organization_id)
db.session.execute(stmt)
db.session.bulk_save_objects(
[
Domain(domain=domain.lower(), organization_id=organization_id)
@@ -76,7 +83,7 @@ def dao_update_organization(organization_id, **kwargs):
]
)
organization = Organization.query.get(organization_id)
organization = db.session.get(Organization, organization_id)
if "organization_type" in kwargs:
_update_organization_services(
organization, "organization_type", only_where_none=False
@@ -101,7 +108,8 @@ def _update_organization_services(organization, attribute, only_where_none=True)
@autocommit
@version_class(Service)
def dao_add_service_to_organization(service, organization_id):
organization = Organization.query.filter_by(id=organization_id).one()
stmt = select(Organization).filter_by(id=organization_id)
organization = db.session.execute(stmt).scalars().one()
service.organization_id = organization_id
service.organization_type = organization.organization_type
@@ -122,7 +130,8 @@ def dao_get_users_for_organization(organization_id):
@autocommit
def dao_add_user_to_organization(organization_id, user_id):
organization = dao_get_organization_by_id(organization_id)
user = User.query.filter_by(id=user_id).one()
stmt = select(User).filter_by(id=user_id)
user = db.session.execute(stmt).scalars().one()
user.organizations.append(organization)
db.session.add(organization)
return user

16
app/dao/service_permissions_dao.py
View File

@@ -1,12 +1,14 @@
from sqlalchemy import delete, select
from app import db
from app.dao.dao_utils import autocommit
from app.models import ServicePermission
def dao_fetch_service_permissions(service_id):
return ServicePermission.query.filter(
ServicePermission.service_id == service_id
).all()
stmt = select(ServicePermission).filter(ServicePermission.service_id == service_id)
return db.session.execute(stmt).scalars().all()
@autocommit
@@ -16,9 +18,11 @@ def dao_add_service_permission(service_id, permission):
def dao_remove_service_permission(service_id, permission):
deleted = ServicePermission.query.filter(
stmt = delete(ServicePermission).where(
ServicePermission.service_id == service_id,
ServicePermission.permission == permission,
).delete()
)
result = db.session.execute(stmt)
db.session.commit()
return deleted
return result.rowcount

15
app/dao/service_sms_sender_dao.py
View File

@@ -1,4 +1,4 @@
from sqlalchemy import desc
from sqlalchemy import desc, select
from app import db
from app.dao.dao_utils import autocommit
@@ -17,17 +17,20 @@ def insert_service_sms_sender(service, sms_sender):
def dao_get_service_sms_senders_by_id(service_id, service_sms_sender_id):
return ServiceSmsSender.query.filter_by(
stmt = select(ServiceSmsSender).filter_by(
id=service_sms_sender_id, service_id=service_id, archived=False
).one()
)
return db.session.execute(stmt).scalars().one()
def dao_get_sms_senders_by_service_id(service_id):
return (
ServiceSmsSender.query.filter_by(service_id=service_id, archived=False)
stmt = (
select(ServiceSmsSender)
.filter_by(service_id=service_id, archived=False)
.order_by(desc(ServiceSmsSender.is_default))
.all()
)
return db.session.execute(stmt).scalars().all()
@autocommit

18
app/dao/service_user_dao.py
View File

@@ -1,25 +1,23 @@
from sqlalchemy import select
from app import db
from app.dao.dao_utils import autocommit
from app.models import ServiceUser, User
def dao_get_service_user(user_id, service_id):
# TODO: This has been changed to account for the test case failure
# that used this method but have any service user to return. Somehow, this
# started to throw an error with one() method in sqlalchemy 2.0 unlike 1.4
return ServiceUser.query.filter_by(
user_id=user_id, service_id=service_id
).one_or_none()
stmt = select(ServiceUser).filter_by(user_id=user_id, service_id=service_id)
return db.session.execute(stmt).scalars().one_or_none()
def dao_get_active_service_users(service_id):
query = (
db.session.query(ServiceUser)
stmt = (
select(ServiceUser)
.join(User, User.id == ServiceUser.user_id)
.filter(User.state == "active", ServiceUser.service_id == service_id)
)
return query.all()
return db.session.execute(stmt).scalars().all()
def dao_get_service_users_by_user_id(user_id):

238
app/dao/services_dao.py
View File

@@ -2,7 +2,7 @@ import uuid
from datetime import timedelta
from flask import current_app
from sqlalchemy import Float, cast, select
from sqlalchemy import Float, cast, delete, select
from sqlalchemy.orm import joinedload
from sqlalchemy.sql.expression import and_, asc, case, func
@@ -51,34 +51,42 @@ from app.utils import (
def dao_fetch_all_services(only_active=False):
query = Service.query.order_by(asc(Service.created_at)).options(
joinedload(Service.users)
)
stmt = select(Service)
if only_active:
query = query.filter(Service.active)
stmt = stmt.where(Service.active)
return query.all()
stmt = stmt.order_by(asc(Service.created_at)).options(joinedload(Service.users))
result = db.session.execute(stmt)
return result.unique().scalars().all()
def get_services_by_partial_name(service_name):
service_name = escape_special_characters(service_name)
return Service.query.filter(Service.name.ilike("%{}%".format(service_name))).all()
stmt = select(Service).where(Service.name.ilike("%{}%".format(service_name)))
result = db.session.execute(stmt)
return result.scalars().all()
def dao_count_live_services():
return Service.query.filter_by(
active=True,
restricted=False,
count_as_live=True,
).count()
stmt = (
select(func.count())
.select_from(Service)
.where(
Service.active, Service.count_as_live, Service.restricted == False # noqa
)
)
result = db.session.execute(stmt)
return result.scalar() # Retrieves the count
def dao_fetch_live_services_data():
year_start_date, year_end_date = get_current_calendar_year()
most_recent_annual_billing = (
db.session.query(
select(
AnnualBilling.service_id,
func.max(AnnualBilling.financial_year_start).label("year"),
)
@@ -86,13 +94,17 @@ def dao_fetch_live_services_data():
.subquery()
)
this_year_ft_billing = FactBilling.query.filter(
FactBilling.local_date >= year_start_date,
FactBilling.local_date <= year_end_date,
).subquery()
this_year_ft_billing = (
select(FactBilling)
.filter(
FactBilling.local_date >= year_start_date,
FactBilling.local_date <= year_end_date,
)
.subquery()
)
data = (
db.session.query(
stmt = (
select(
Service.id.label("service_id"),
Service.name.label("service_name"),
Organization.name.label("organization_name"),
@@ -156,8 +168,9 @@ def dao_fetch_live_services_data():
AnnualBilling.free_sms_fragment_limit,
)
.order_by(asc(Service.go_live_at))
.all()
)
data = db.session.execute(stmt).all()
results = []
for row in data:
existing_service = next(
@@ -183,48 +196,55 @@ def dao_fetch_service_by_id(service_id, only_active=False):
stmt = stmt.where(Service.active)
result = db.session.execute(stmt)
return result.unique().scalars().one()
return result.unique().scalars().unique().one()
def dao_fetch_service_by_inbound_number(number):
inbound_number = InboundNumber.query.filter(
stmt = select(InboundNumber).where(
InboundNumber.number == number, InboundNumber.active
).first()
)
result = db.session.execute(stmt)
inbound_number = result.scalars().first()
if not inbound_number:
return None
return Service.query.filter(Service.id == inbound_number.service_id).first()
stmt = select(Service).where(Service.id == inbound_number.service_id)
result = db.session.execute(stmt)
return result.scalars().first()
def dao_fetch_service_by_id_with_api_keys(service_id, only_active=False):
query = Service.query.filter_by(id=service_id).options(joinedload(Service.api_keys))
stmt = (
select(Service).filter_by(id=service_id).options(joinedload(Service.api_keys))
)
if only_active:
query = query.filter(Service.active)
return query.one()
stmt = stmt.filter(Service.active)
return db.session.execute(stmt).scalars().unique().one()
def dao_fetch_all_services_by_user(user_id, only_active=False):
query = (
Service.query.filter(Service.users.any(id=user_id))
stmt = (
select(Service)
.filter(Service.users.any(id=user_id))
.order_by(asc(Service.created_at))
.options(joinedload(Service.users))
)
if only_active:
query = query.filter(Service.active)
return query.all()
stmt = stmt.filter(Service.active)
return db.session.execute(stmt).scalars().unique().all()
def dao_fetch_all_services_created_by_user(user_id):
query = Service.query.filter_by(created_by_id=user_id).order_by(
asc(Service.created_at)
stmt = (
select(Service)
.filter_by(created_by_id=user_id)
.order_by(asc(Service.created_at))
)
return query.all()
return db.session.execute(stmt).scalars().all()
@autocommit
@@ -234,16 +254,15 @@ def dao_fetch_all_services_created_by_user(user_id):
VersionOptions(Template, history_class=TemplateHistory, must_write_history=False),
)
def dao_archive_service(service_id):
# have to eager load templates and api keys so that we don't flush when we loop through them
# to ensure that db.session still contains the models when it comes to creating history objects
service = (
Service.query.options(
stmt = (
select(Service)
.options(
joinedload(Service.templates).subqueryload(Template.template_redacted),
joinedload(Service.api_keys),
)
.filter(Service.id == service_id)
.one()
)
service = db.session.execute(stmt).scalars().unique().one()
service.active = False
service.name = get_archived_db_column_value(service.name)
@@ -259,11 +278,14 @@ def dao_archive_service(service_id):
def dao_fetch_service_by_id_and_user(service_id, user_id):
return (
Service.query.filter(Service.users.any(id=user_id), Service.id == service_id)
stmt = (
select(Service)
.filter(Service.users.any(id=user_id), Service.id == service_id)
.options(joinedload(Service.users))
.one()
)
result = db.session.execute(stmt).scalar_one()
return result
@autocommit
@@ -366,39 +388,40 @@ def dao_remove_user_from_service(service, user):
def delete_service_and_all_associated_db_objects(service):
def _delete_commit(query):
query.delete(synchronize_session=False)
def _delete_commit(stmt):
db.session.execute(stmt)
db.session.commit()
subq = db.session.query(Template.id).filter_by(service=service).subquery()
_delete_commit(
TemplateRedacted.query.filter(TemplateRedacted.template_id.in_(subq))
)
subq = select(Template.id).filter_by(service=service).subquery()
_delete_commit(ServiceSmsSender.query.filter_by(service=service))
_delete_commit(ServiceEmailReplyTo.query.filter_by(service=service))
_delete_commit(InvitedUser.query.filter_by(service=service))
_delete_commit(Permission.query.filter_by(service=service))
_delete_commit(NotificationHistory.query.filter_by(service=service))
_delete_commit(Notification.query.filter_by(service=service))
_delete_commit(Job.query.filter_by(service=service))
_delete_commit(Template.query.filter_by(service=service))
_delete_commit(TemplateHistory.query.filter_by(service_id=service.id))
_delete_commit(ServicePermission.query.filter_by(service_id=service.id))
_delete_commit(ApiKey.query.filter_by(service=service))
_delete_commit(ApiKey.get_history_model().query.filter_by(service_id=service.id))
_delete_commit(AnnualBilling.query.filter_by(service_id=service.id))
stmt = delete(TemplateRedacted).filter(TemplateRedacted.template_id.in_(subq))
_delete_commit(stmt)
verify_codes = VerifyCode.query.join(User).filter(
User.id.in_([x.id for x in service.users])
_delete_commit(delete(ServiceSmsSender).filter_by(service=service))
_delete_commit(delete(ServiceEmailReplyTo).filter_by(service=service))
_delete_commit(delete(InvitedUser).filter_by(service=service))
_delete_commit(delete(Permission).filter_by(service=service))
_delete_commit(delete(NotificationHistory).filter_by(service=service))
_delete_commit(delete(Notification).filter_by(service=service))
_delete_commit(delete(Job).filter_by(service=service))
_delete_commit(delete(Template).filter_by(service=service))
_delete_commit(delete(TemplateHistory).filter_by(service_id=service.id))
_delete_commit(delete(ServicePermission).filter_by(service_id=service.id))
_delete_commit(delete(ApiKey).filter_by(service=service))
_delete_commit(delete(ApiKey.get_history_model()).filter_by(service_id=service.id))
_delete_commit(delete(AnnualBilling).filter_by(service_id=service.id))
stmt = (
select(VerifyCode).join(User).filter(User.id.in_([x.id for x in service.users]))
)
verify_codes = db.session.execute(stmt).scalars().all()
list(map(db.session.delete, verify_codes))
db.session.commit()
users = [x for x in service.users]
for user in users:
user.organizations = []
service.users.remove(user)
_delete_commit(Service.get_history_model().query.filter_by(id=service.id))
_delete_commit(delete(Service.get_history_model()).filter_by(id=service.id))
db.session.delete(service)
db.session.commit()
for user in users:
@@ -409,8 +432,8 @@ def delete_service_and_all_associated_db_objects(service):
def dao_fetch_todays_stats_for_service(service_id):
today = utc_now().date()
start_date = get_midnight_in_utc(today)
return (
db.session.query(
stmt = (
select(
Notification.notification_type,
Notification.status,
func.count(Notification.id).label("count"),
@@ -424,16 +447,16 @@ def dao_fetch_todays_stats_for_service(service_id):
Notification.notification_type,
Notification.status,
)
.all()
)
return db.session.execute(stmt).all()
def dao_fetch_stats_for_service_from_days(service_id, start_date, end_date):
start_date = get_midnight_in_utc(start_date)
end_date = get_midnight_in_utc(end_date + timedelta(days=1))
return (
db.session.query(
stmt = (
select(
NotificationAllTimeView.notification_type,
NotificationAllTimeView.status,
func.date_trunc("day", NotificationAllTimeView.created_at).label("day"),
@@ -450,8 +473,8 @@ def dao_fetch_stats_for_service_from_days(service_id, start_date, end_date):
NotificationAllTimeView.status,
func.date_trunc("day", NotificationAllTimeView.created_at),
)
.all()
)
return db.session.execute(stmt).scalars().all()
def dao_fetch_stats_for_service_from_days_for_user(
@@ -460,13 +483,14 @@ def dao_fetch_stats_for_service_from_days_for_user(
start_date = get_midnight_in_utc(start_date)
end_date = get_midnight_in_utc(end_date + timedelta(days=1))
return (
db.session.query(
stmt = (
select(
NotificationAllTimeView.notification_type,
NotificationAllTimeView.status,
func.date_trunc("day", NotificationAllTimeView.created_at).label("day"),
func.count(NotificationAllTimeView.id).label("count"),
)
.select_from(NotificationAllTimeView)
.filter(
NotificationAllTimeView.service_id == service_id,
NotificationAllTimeView.key_type != KeyType.TEST,
@@ -479,8 +503,8 @@ def dao_fetch_stats_for_service_from_days_for_user(
NotificationAllTimeView.status,
func.date_trunc("day", NotificationAllTimeView.created_at),
)
.all()
)
return db.session.execute(stmt).scalars().all()
def dao_fetch_todays_stats_for_all_services(
@@ -491,7 +515,7 @@ def dao_fetch_todays_stats_for_all_services(
end_date = get_midnight_in_utc(today + timedelta(days=1))
subquery = (
db.session.query(
select(
Notification.notification_type,
Notification.status,
Notification.service_id,
@@ -510,8 +534,8 @@ def dao_fetch_todays_stats_for_all_services(
subquery = subquery.subquery()
query = (
db.session.query(
stmt = (
select(
Service.id.label("service_id"),
Service.name,
Service.restricted,
@@ -526,9 +550,9 @@ def dao_fetch_todays_stats_for_all_services(
)
if only_active:
query = query.filter(Service.active)
stmt = stmt.filter(Service.active)
return query.all()
return db.session.execute(stmt).all()
@autocommit
@@ -537,15 +561,13 @@ def dao_fetch_todays_stats_for_all_services(
VersionOptions(Service),
)
def dao_suspend_service(service_id):
# have to eager load api keys so that we don't flush when we loop through them
# to ensure that db.session still contains the models when it comes to creating history objects
service = (
Service.query.options(
joinedload(Service.api_keys),
)
stmt = (
select(Service)
.options(joinedload(Service.api_keys))
.filter(Service.id == service_id)
.one()
)
service = db.session.execute(stmt).scalars().unique().one()
for api_key in service.api_keys:
if not api_key.expiry_date:
@@ -557,19 +579,22 @@ def dao_suspend_service(service_id):
@autocommit
@version_class(Service)
def dao_resume_service(service_id):
service = Service.query.get(service_id)
service = db.session.get(Service, service_id)
service.active = True
def dao_fetch_active_users_for_service(service_id):
query = User.query.filter(User.services.any(id=service_id), User.state == "active")
return query.all()
stmt = select(User).where(User.services.any(id=service_id), User.state == "active")
result = db.session.execute(stmt)
return result.scalars().all()
def dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=500):
return (
db.session.query(
stmt = (
select(
Notification.service_id.label("service_id"),
func.count(Notification.id).label("notification_count"),
)
@@ -587,13 +612,13 @@ def dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=500)
Notification.service_id,
)
.having(func.count(Notification.id) > threshold)
.all()
)
return db.session.execute(stmt).all()
def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10000):
subquery = (
db.session.query(
select(
func.count(Notification.id).label("total_count"),
Notification.service_id.label("service_id"),
)
@@ -614,8 +639,8 @@ def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10
subquery = subquery.subquery()
query = (
db.session.query(
stmt = (
select(
Notification.service_id.label("service_id"),
func.count(Notification.id).label("permanent_failure_count"),
subquery.c.total_count.label("total_count"),
@@ -643,17 +668,19 @@ def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10
)
)
return query.all()
return db.session.execute(stmt).all()
def get_live_services_with_organization():
query = (
db.session.query(
stmt = (
select(
Service.id.label("service_id"),
Service.name.label("service_name"),
Organization.id.label("organization_id"),
Organization.name.label("organization_name"),
)
.select_from(Service)
.outerjoin(Service.organization)
.filter(
Service.count_as_live.is_(True),
@@ -663,14 +690,15 @@ def get_live_services_with_organization():
.order_by(Organization.name, Service.name)
)
return query.all()
return db.session.execute(stmt).all()
def fetch_notification_stats_for_service_by_month_by_user(
start_date, end_date, service_id, user_id
):
return (
db.session.query(
stmt = (
select(
func.date_trunc("month", NotificationAllTimeView.created_at).label("month"),
NotificationAllTimeView.notification_type,
(NotificationAllTimeView.status).label("notification_status"),
@@ -688,8 +716,8 @@ def fetch_notification_stats_for_service_by_month_by_user(
NotificationAllTimeView.notification_type,
NotificationAllTimeView.status,
)
.all()
)
return db.session.execute(stmt).all()
def get_specific_days_stats(data, start_date, days=None, end_date=None):

10
app/dao/template_folder_dao.py
View File

@@ -1,16 +1,20 @@
from sqlalchemy import select
from app import db
from app.dao.dao_utils import autocommit
from app.models import TemplateFolder
def dao_get_template_folder_by_id_and_service_id(template_folder_id, service_id):
return TemplateFolder.query.filter(
stmt = select(TemplateFolder).filter(
TemplateFolder.id == template_folder_id, TemplateFolder.service_id == service_id
).one()
)
return db.session.execute(stmt).scalars().one()
def dao_get_valid_template_folders_by_id(folder_ids):
return TemplateFolder.query.filter(TemplateFolder.id.in_(folder_ids)).all()
stmt = select(TemplateFolder).filter(TemplateFolder.id.in_(folder_ids))
return db.session.execute(stmt).scalars().all()
@autocommit

40
app/dao/templates_dao.py
View File

@@ -1,6 +1,6 @@
import uuid
from sqlalchemy import asc, desc
from sqlalchemy import asc, desc, select
from app import db
from app.dao.dao_utils import VersionOptions, autocommit, version_class
@@ -46,24 +46,29 @@ def dao_redact_template(template, user_id):
def dao_get_template_by_id_and_service_id(template_id, service_id, version=None):
if version is not None:
return TemplateHistory.query.filter_by(
stmt = select(TemplateHistory).filter_by(
id=template_id, hidden=False, service_id=service_id, version=version
).one()
return Template.query.filter_by(
)
return db.session.execute(stmt).scalars().one()
stmt = select(Template).filter_by(
id=template_id, hidden=False, service_id=service_id
).one()
)
return db.session.execute(stmt).scalars().one()
def dao_get_template_by_id(template_id, version=None):
if version is not None:
return TemplateHistory.query.filter_by(id=template_id, version=version).one()
return Template.query.filter_by(id=template_id).one()
stmt = select(TemplateHistory).filter_by(id=template_id, version=version)
return db.session.execute(stmt).scalars().one()
stmt = select(Template).filter_by(id=template_id)
return db.session.execute(stmt).scalars().one()
def dao_get_all_templates_for_service(service_id, template_type=None):
if template_type is not None:
return (
Template.query.filter_by(
stmt = (
select(Template)
.filter_by(
service_id=service_id,
template_type=template_type,
hidden=False,
@@ -73,26 +78,27 @@ def dao_get_all_templates_for_service(service_id, template_type=None):
asc(Template.name),
asc(Template.template_type),
)
.all()
)
return (
Template.query.filter_by(service_id=service_id, hidden=False, archived=False)
return db.session.execute(stmt).scalars().all()
stmt = (
select(Template)
.filter_by(service_id=service_id, hidden=False, archived=False)
.order_by(
asc(Template.name),
asc(Template.template_type),
)
.all()
)
return db.session.execute(stmt).scalars().all()
def dao_get_template_versions(service_id, template_id):
return (
TemplateHistory.query.filter_by(
stmt = (
select(TemplateHistory)
.filter_by(
service_id=service_id,
id=template_id,
hidden=False,
)
.order_by(desc(TemplateHistory.version))
.all()
)
return db.session.execute(stmt).scalars().all()

55
app/dao/users_dao.py
View File

@@ -4,7 +4,7 @@ from secrets import randbelow
import sqlalchemy
from flask import current_app
from sqlalchemy import func, text
from sqlalchemy import delete, func, select, text
from sqlalchemy.orm import joinedload
from app import db
@@ -37,8 +37,8 @@ def get_login_gov_user(login_uuid, email_address):
login.gov uuids are. Eventually the code that checks by email address
should be removed.
"""
user = User.query.filter_by(login_uuid=login_uuid).first()
stmt = select(User).filter_by(login_uuid=login_uuid)
user = db.session.execute(stmt).scalars().first()
if user:
if user.email_address != email_address:
try:
@@ -54,7 +54,8 @@ def get_login_gov_user(login_uuid, email_address):
return user
# Remove this 1 July 2025, all users should have login.gov uuids by now
user = User.query.filter(User.email_address.ilike(email_address)).first()
stmt = select(User).filter(User.email_address.ilike(email_address))
user = db.session.execute(stmt).scalars().first()
if user:
save_user_attribute(user, {"login_uuid": login_uuid})
@@ -102,24 +103,27 @@ def create_user_code(user, code, code_type):
def get_user_code(user, code, code_type):
# Get the most recent codes to try and reduce the
# time searching for the correct code.
codes = VerifyCode.query.filter_by(user=user, code_type=code_type).order_by(
VerifyCode.created_at.desc()
stmt = (
select(VerifyCode)
.filter_by(user=user, code_type=code_type)
.order_by(VerifyCode.created_at.desc())
)
codes = db.session.execute(stmt).scalars().all()
return next((x for x in codes if x.check_code(code)), None)
def delete_codes_older_created_more_than_a_day_ago():
deleted = (
db.session.query(VerifyCode)
.filter(VerifyCode.created_at < utc_now() - timedelta(hours=24))
.delete()
stmt = delete(VerifyCode).filter(
VerifyCode.created_at < utc_now() - timedelta(hours=24)
)
deleted = db.session.execute(stmt)
db.session.commit()
return deleted
def use_user_code(id):
verify_code = VerifyCode.query.get(id)
verify_code = db.session.get(VerifyCode, id)
verify_code.code_used = True
db.session.add(verify_code)
db.session.commit()
@@ -131,36 +135,42 @@ def delete_model_user(user):
def delete_user_verify_codes(user):
VerifyCode.query.filter_by(user=user).delete()
stmt = delete(VerifyCode).filter_by(user=user)
db.session.execute(stmt)
db.session.commit()
def count_user_verify_codes(user):
query = VerifyCode.query.filter(
stmt = select(func.count(VerifyCode.id)).filter(
VerifyCode.user == user,
VerifyCode.expiry_datetime > utc_now(),
VerifyCode.code_used.is_(False),
)
return query.count()
result = db.session.execute(stmt).scalar()
return result or 0
def get_user_by_id(user_id=None):
if user_id:
return User.query.filter_by(id=user_id).one()
return User.query.filter_by().all()
stmt = select(User).filter_by(id=user_id)
return db.session.execute(stmt).scalars().one()
return get_users()
def get_users():
return User.query.all()
stmt = select(User)
return db.session.execute(stmt).scalars().all()
def get_user_by_email(email):
return User.query.filter(func.lower(User.email_address) == func.lower(email)).one()
stmt = select(User).filter(func.lower(User.email_address) == func.lower(email))
return db.session.execute(stmt).scalars().one()
def get_users_by_partial_email(email):
email = escape_special_characters(email)
return User.query.filter(User.email_address.ilike("%{}%".format(email))).all()
stmt = select(User).filter(User.email_address.ilike("%{}%".format(email)))
return db.session.execute(stmt).scalars().all()
def increment_failed_login_count(user):
@@ -188,16 +198,17 @@ def get_user_and_accounts(user_id):
# TODO: With sqlalchemy 2.0 change as below because of the breaking change
# at User.organizations.services, we need to verify that the below subqueryload
# that we have put is functionally doing the same thing as before
return (
User.query.filter(User.id == user_id)
stmt = (
select(User)
.filter(User.id == user_id)
.options(
# eagerly load the user's services and organizations, and also the service's org and vice versa
# (so we can see if the user knows about it)
joinedload(User.services).joinedload(Service.organization),
joinedload(User.organizations).subqueryload(Organization.services),
)
.one()
)
return db.session.execute(stmt).scalars().unique().one()
@autocommit

24
app/delivery/send_to_providers.py
View File

@@ -98,17 +98,7 @@ def send_sms_to_provider(notification):
# TODO This is temporary to test the capability of validating phone numbers
# The future home of the validation is TBD
if "+" not in recipient:
recipient_lookup = f"+{recipient}"
else:
recipient_lookup = recipient
if recipient_lookup in current_app.config[
"SIMULATED_SMS_NUMBERS"
] and os.getenv("NOTIFY_ENVIRONMENT") in ["development", "test"]:
current_app.logger.info(hilite("#validate-phone-number fired"))
aws_pinpoint_client.validate_phone_number("01", recipient)
else:
current_app.logger.info(hilite("#validate-phone-number not fired"))
_experimentally_validate_phone_numbers(recipient)
sender_numbers = get_sender_numbers(notification)
if notification.reply_to_text not in sender_numbers:
@@ -145,6 +135,18 @@ def send_sms_to_provider(notification):
return message_id
def _experimentally_validate_phone_numbers(recipient):
if "+" not in recipient:
recipient_lookup = f"+{recipient}"
else:
recipient_lookup = recipient
if recipient_lookup in current_app.config["SIMULATED_SMS_NUMBERS"] and os.getenv(
"NOTIFY_ENVIRONMENT"
) in ["development", "test"]:
current_app.logger.info(hilite("#validate-phone-number fired"))
aws_pinpoint_client.validate_phone_number("01", recipient)
def _get_verify_code(notification):
key = f"2facode-{notification.id}".replace(" ", "")
recipient = redis_store.get(key)

57
app/service/rest.py
View File

@@ -453,16 +453,6 @@ def get_all_notifications_for_service(service_id):
data = notifications_filter_schema.load(MultiDict(request.get_json()))
current_app.logger.debug(f"use POST, request {request.get_json()} data {data}")
if data.get("to"):
notification_type = (
data.get("template_type")[0] if data.get("template_type") else None
)
return search_for_notification_by_to_field(
service_id=service_id,
search_term=data["to"],
statuses=data.get("status"),
notification_type=notification_type,
)
page = data["page"] if "page" in data else 1
page_size = (
data["page_size"]
@@ -583,53 +573,6 @@ def get_notification_for_service(service_id, notification_id):
)
def search_for_notification_by_to_field(
service_id, search_term, statuses, notification_type
):
results = notifications_dao.dao_get_notifications_by_recipient_or_reference(
service_id=service_id,
search_term=search_term,
statuses=statuses,
notification_type=notification_type,
page=1,
page_size=current_app.config["PAGE_SIZE"],
)
# 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. Note, this was previously be done by having
# notifications_dao.dao_get_notifications_by_recipient_or_reference use count=False when calling
# Flask-Sqlalchemys `paginate'. But instead we now use this way because it is much more performant for
# services with many results (unlike using Flask SqlAlchemy `paginate` with `count=True`, 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.dao_get_notifications_by_recipient_or_reference(
service_id=service_id,
search_term=search_term,
statuses=statuses,
notification_type=notification_type,
page=2,
page_size=current_app.config["PAGE_SIZE"],
error_out=False, # False so that if there are no results, it doesn't end in aborting with a 404
)
return (
jsonify(
notifications=notification_with_template_schema.dump(
results.items, many=True
),
links=get_prev_next_pagination_links(
1,
len(next_page_of_pagination.items),
".get_all_notifications_for_service",
statuses=statuses,
notification_type=notification_type,
service_id=service_id,
),
),
200,
)
@service_blueprint.route("/<uuid:service_id>/notifications/monthly", methods=["GET"])
def get_monthly_notification_stats(service_id):
# check service_id validity

19
app/service_invite/rest.py
View File

@@ -32,7 +32,7 @@ service_invite = Blueprint("service_invite", __name__)
register_errors(service_invite)
def _create_service_invite(invited_user, invite_link_host):
def _create_service_invite(invited_user, nonce):
template_id = current_app.config["INVITATION_EMAIL_TEMPLATE_ID"]
@@ -40,12 +40,6 @@ def _create_service_invite(invited_user, invite_link_host):
service = Service.query.get(current_app.config["NOTIFY_SERVICE_ID"])
token = generate_token(
str(invited_user.email_address),
current_app.config["SECRET_KEY"],
current_app.config["DANGEROUS_SALT"],
)
# The raw permissions are in the form "a,b,c,d"
# but need to be in the form ["a", "b", "c", "d"]
data = {}
@@ -59,7 +53,8 @@ def _create_service_invite(invited_user, invite_link_host):
data["invited_user_email"] = invited_user.email_address
url = os.environ["LOGIN_DOT_GOV_REGISTRATION_URL"]
url = url.replace("NONCE", token)
url = url.replace("NONCE", nonce) # handed from data sent from admin.
user_data_url_safe = get_user_data_url_safe(data)
@@ -94,10 +89,16 @@ def _create_service_invite(invited_user, invite_link_host):
@service_invite.route("/service/<service_id>/invite", methods=["POST"])
def create_invited_user(service_id):
request_json = request.get_json()
try:
nonce = request_json.pop("nonce")
except KeyError:
current_app.logger.exception("nonce not found in submitted data.")
raise
invited_user = invited_user_schema.load(request_json)
save_invited_user(invited_user)
_create_service_invite(invited_user, request_json.get("invite_link_host"))
_create_service_invite(invited_user, nonce)
return jsonify(data=invited_user_schema.dump(invited_user)), 201