darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-09-06 23:18:25 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into 2199-add-pending-message-data-to-daily-and-user_daily-stats

Browse Source
This commit is contained in:
Beverly Nguyen
2025-01-21 16:40:19 -08:00
parent 880237f55d e6899ef9d0
commit 1fc78d81e5
98 changed files with 1678 additions and 743 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
app/dao/annual_billing_dao.py
View File

@@ -29,8 +29,8 @@ def dao_create_or_update_annual_billing_for_year(
def dao_get_annual_billing(service_id):
stmt = (
select(AnnualBilling)
.filter_by(
service_id=service_id,
.where(
AnnualBilling.service_id == service_id,
)
.order_by(AnnualBilling.financial_year_start)
)
@@ -43,7 +43,7 @@ def dao_update_annual_billing_for_future_years(
):
stmt = (
update(AnnualBilling)
.filter(
.where(
AnnualBilling.service_id == service_id,
AnnualBilling.financial_year_start > financial_year_start,
)
@@ -57,8 +57,9 @@ def dao_get_free_sms_fragment_limit_for_year(service_id, financial_year_start=No
if not financial_year_start:
financial_year_start = get_current_calendar_year_start_year()
stmt = select(AnnualBilling).filter_by(
service_id=service_id, financial_year_start=financial_year_start
stmt = select(AnnualBilling).where(
AnnualBilling.service_id == service_id,
AnnualBilling.financial_year_start == financial_year_start,
)
return db.session.execute(stmt).scalars().first()
@@ -66,8 +67,8 @@ def dao_get_free_sms_fragment_limit_for_year(service_id, financial_year_start=No
def dao_get_all_free_sms_fragment_limit(service_id):
stmt = (
select(AnnualBilling)
.filter_by(
service_id=service_id,
.where(
AnnualBilling.service_id == service_id,
)
.order_by(AnnualBilling.financial_year_start)
)

66
app/dao/api_key_dao.py
View File

@@ -1,7 +1,7 @@
import uuid
from datetime import timedelta
from sqlalchemy import func, or_
from sqlalchemy import func, or_, select
from app import db
from app.dao.dao_utils import autocommit, version_class
@@ -23,31 +23,61 @@ def save_model_api_key(api_key):
@autocommit
@version_class(ApiKey)
def expire_api_key(service_id, api_key_id):
api_key = ApiKey.query.filter_by(id=api_key_id, service_id=service_id).one()
api_key = (
db.session.execute(
select(ApiKey).where(
ApiKey.id == api_key_id, ApiKey.service_id == service_id
)
)
.scalars()
.one()
)
api_key.expiry_date = utc_now()
db.session.add(api_key)
def get_model_api_keys(service_id, id=None):
if id:
return ApiKey.query.filter_by(
id=id, service_id=service_id, expiry_date=None
).one()
return (
db.session.execute(
select(ApiKey).where(
ApiKey.id == id,
ApiKey.service_id == service_id,
ApiKey.expiry_date == None, # noqa
)
)
.scalars()
.one()
)
seven_days_ago = utc_now() - timedelta(days=7)
return ApiKey.query.filter(
or_(
ApiKey.expiry_date == None, # noqa
func.date(ApiKey.expiry_date) > seven_days_ago, # noqa
),
ApiKey.service_id == service_id,
).all()
return (
db.session.execute(
select(ApiKey).where(
or_(
ApiKey.expiry_date == None, # noqa
func.date(ApiKey.expiry_date) > seven_days_ago, # noqa
),
ApiKey.service_id == service_id,
)
)
.scalars()
.all()
)
def get_unsigned_secrets(service_id):
"""
This method can only be exposed to the Authentication of the api calls.
"""
api_keys = ApiKey.query.filter_by(service_id=service_id, expiry_date=None).all()
api_keys = (
db.session.execute(
select(ApiKey).where(
ApiKey.service_id == service_id, ApiKey.expiry_date == None # noqa
)
)
.scalars()
.all()
)
keys = [x.secret for x in api_keys]
return keys
@@ -56,5 +86,13 @@ def get_unsigned_secret(key_id):
"""
This method can only be exposed to the Authentication of the api calls.
"""
api_key = ApiKey.query.filter_by(id=key_id, expiry_date=None).one()
api_key = (
db.session.execute(
select(ApiKey).where(
ApiKey.id == key_id, ApiKey.expiry_date == None # noqa
)
)
.scalars()
.one()
)
return api_key.secret

4
app/dao/complaint_dao.py
View File

@@ -33,7 +33,7 @@ def fetch_paginated_complaints(page=1):
def fetch_complaints_by_service(service_id):
stmt = (
select(Complaint)
.filter_by(service_id=service_id)
.where(Complaint.service_id == service_id)
.order_by(desc(Complaint.created_at))
)
return db.session.execute(stmt).scalars().all()
@@ -46,6 +46,6 @@ def fetch_count_of_complaints(start_date, end_date):
stmt = (
select(func.count())
.select_from(Complaint)
.filter(Complaint.created_at >= start_date, Complaint.created_at < end_date)
.where(Complaint.created_at >= start_date, Complaint.created_at < end_date)
)
return db.session.execute(stmt).scalar() or 0

20
app/dao/email_branding_dao.py
View File

@@ -1,18 +1,32 @@
from sqlalchemy import select
from app import db
from app.dao.dao_utils import autocommit
from app.models import EmailBranding
def dao_get_email_branding_options():
return EmailBranding.query.all()
return db.session.execute(select(EmailBranding)).scalars().all()
def dao_get_email_branding_by_id(email_branding_id):
return EmailBranding.query.filter_by(id=email_branding_id).one()
return (
db.session.execute(
select(EmailBranding).where(EmailBranding.id == email_branding_id)
)
.scalars()
.one()
)
def dao_get_email_branding_by_name(email_branding_name):
return EmailBranding.query.filter_by(name=email_branding_name).first()
return (
db.session.execute(
select(EmailBranding).where(EmailBranding.name == email_branding_name)
)
.scalars()
.first()
)
@autocommit

106
app/dao/fact_billing_dao.py
View File

@@ -52,7 +52,7 @@ def fetch_sms_free_allowance_remainder_until_date(end_date):
FactBilling.notification_type == NotificationType.SMS,
),
)
.filter(
.where(
AnnualBilling.financial_year_start == billing_year,
)
.group_by(
@@ -65,7 +65,7 @@ def fetch_sms_free_allowance_remainder_until_date(end_date):
def fetch_sms_billing_for_all_services(start_date, end_date):
# ASSUMPTION: AnnualBilling has been populated for year.
allowance_left_at_start_date_query = fetch_sms_free_allowance_remainder_until_date(
allowance_left_at_start_date_stmt = fetch_sms_free_allowance_remainder_until_date(
start_date
).subquery()
@@ -76,14 +76,14 @@ def fetch_sms_billing_for_all_services(start_date, end_date):
# subtract sms_billable_units units accrued since report's start date to get up-to-date
# allowance remainder
sms_allowance_left = func.greatest(
allowance_left_at_start_date_query.c.sms_remainder - sms_billable_units, 0
allowance_left_at_start_date_stmt.c.sms_remainder - sms_billable_units, 0
)
# billable units here are for period between start date and end date only, so to see
# how many are chargeable, we need to see how much free allowance was used up in the
# period up until report's start date and then do a subtraction
chargeable_sms = func.greatest(
sms_billable_units - allowance_left_at_start_date_query.c.sms_remainder, 0
sms_billable_units - allowance_left_at_start_date_stmt.c.sms_remainder, 0
)
sms_cost = chargeable_sms * FactBilling.rate
@@ -93,7 +93,7 @@ def fetch_sms_billing_for_all_services(start_date, end_date):
Organization.id.label("organization_id"),
Service.name.label("service_name"),
Service.id.label("service_id"),
allowance_left_at_start_date_query.c.free_sms_fragment_limit,
allowance_left_at_start_date_stmt.c.free_sms_fragment_limit,
FactBilling.rate.label("sms_rate"),
sms_allowance_left.label("sms_remainder"),
sms_billable_units.label("sms_billable_units"),
@@ -102,15 +102,15 @@ def fetch_sms_billing_for_all_services(start_date, end_date):
)
.select_from(Service)
.outerjoin(
allowance_left_at_start_date_query,
Service.id == allowance_left_at_start_date_query.c.service_id,
allowance_left_at_start_date_stmt,
Service.id == allowance_left_at_start_date_stmt.c.service_id,
)
.outerjoin(Service.organization)
.join(
FactBilling,
FactBilling.service_id == Service.id,
)
.filter(
.where(
FactBilling.local_date >= start_date,
FactBilling.local_date <= end_date,
FactBilling.notification_type == NotificationType.SMS,
@@ -120,8 +120,8 @@ def fetch_sms_billing_for_all_services(start_date, end_date):
Organization.id,
Service.id,
Service.name,
allowance_left_at_start_date_query.c.free_sms_fragment_limit,
allowance_left_at_start_date_query.c.sms_remainder,
allowance_left_at_start_date_stmt.c.free_sms_fragment_limit,
allowance_left_at_start_date_stmt.c.sms_remainder,
FactBilling.rate,
)
.order_by(Organization.name, Service.name)
@@ -151,15 +151,15 @@ def fetch_billing_totals_for_year(service_id, year):
union(
*[
select(
query.c.notification_type.label("notification_type"),
query.c.rate.label("rate"),
func.sum(query.c.notifications_sent).label("notifications_sent"),
func.sum(query.c.chargeable_units).label("chargeable_units"),
func.sum(query.c.cost).label("cost"),
func.sum(query.c.free_allowance_used).label("free_allowance_used"),
func.sum(query.c.charged_units).label("charged_units"),
).group_by(query.c.rate, query.c.notification_type)
for query in [
stmt.c.notification_type.label("notification_type"),
stmt.c.rate.label("rate"),
func.sum(stmt.c.notifications_sent).label("notifications_sent"),
func.sum(stmt.c.chargeable_units).label("chargeable_units"),
func.sum(stmt.c.cost).label("cost"),
func.sum(stmt.c.free_allowance_used).label("free_allowance_used"),
func.sum(stmt.c.charged_units).label("charged_units"),
).group_by(stmt.c.rate, stmt.c.notification_type)
for stmt in [
query_service_sms_usage_for_year(service_id, year).subquery(),
query_service_email_usage_for_year(service_id, year).subquery(),
]
@@ -206,22 +206,22 @@ def fetch_monthly_billing_for_year(service_id, year):
union(
*[
select(
query.c.rate.label("rate"),
query.c.notification_type.label("notification_type"),
func.date_trunc("month", query.c.local_date)
stmt.c.rate.label("rate"),
stmt.c.notification_type.label("notification_type"),
func.date_trunc("month", stmt.c.local_date)
.cast(Date)
.label("month"),
func.sum(query.c.notifications_sent).label("notifications_sent"),
func.sum(query.c.chargeable_units).label("chargeable_units"),
func.sum(query.c.cost).label("cost"),
func.sum(query.c.free_allowance_used).label("free_allowance_used"),
func.sum(query.c.charged_units).label("charged_units"),
func.sum(stmt.c.notifications_sent).label("notifications_sent"),
func.sum(stmt.c.chargeable_units).label("chargeable_units"),
func.sum(stmt.c.cost).label("cost"),
func.sum(stmt.c.free_allowance_used).label("free_allowance_used"),
func.sum(stmt.c.charged_units).label("charged_units"),
).group_by(
query.c.rate,
query.c.notification_type,
stmt.c.rate,
stmt.c.notification_type,
"month",
)
for query in [
for stmt in [
query_service_sms_usage_for_year(service_id, year).subquery(),
query_service_email_usage_for_year(service_id, year).subquery(),
]
@@ -250,7 +250,7 @@ def query_service_email_usage_for_year(service_id, year):
FactBilling.billable_units.label("charged_units"),
)
.select_from(FactBilling)
.filter(
.where(
FactBilling.service_id == service_id,
FactBilling.local_date >= year_start,
FactBilling.local_date <= year_end,
@@ -338,7 +338,7 @@ def query_service_sms_usage_for_year(service_id, year):
)
.select_from(FactBilling)
.join(AnnualBilling, AnnualBilling.service_id == service_id)
.filter(
.where(
FactBilling.service_id == service_id,
FactBilling.local_date >= year_start,
FactBilling.local_date <= year_end,
@@ -355,7 +355,7 @@ def delete_billing_data_for_service_for_day(process_day, service_id):
Returns how many rows were deleted
"""
stmt = delete(FactBilling).filter(
stmt = delete(FactBilling).where(
FactBilling.local_date == process_day, FactBilling.service_id == service_id
)
result = db.session.execute(stmt)
@@ -371,9 +371,9 @@ def fetch_billing_data_for_day(process_day, service_id=None, check_permissions=F
)
transit_data = []
if not service_id:
services = Service.query.all()
services = db.session.execute(select(Service)).scalars().all()
else:
services = [Service.query.get(service_id)]
services = [db.session.get(Service, service_id)]
for service in services:
for notification_type in (NotificationType.SMS, NotificationType.EMAIL):
@@ -403,7 +403,7 @@ def _query_for_billing_data(notification_type, start_date, end_date, service):
func.count().label("notifications_sent"),
)
.select_from(NotificationAllTimeView)
.filter(
.where(
NotificationAllTimeView.status.in_(
NotificationStatus.sent_email_types()
),
@@ -438,7 +438,7 @@ def _query_for_billing_data(notification_type, start_date, end_date, service):
func.count().label("notifications_sent"),
)
.select_from(NotificationAllTimeView)
.filter(
.where(
NotificationAllTimeView.status.in_(
NotificationStatus.billable_sms_types()
),
@@ -474,7 +474,7 @@ def get_service_ids_that_need_billing_populated(start_date, end_date):
stmt = (
select(NotificationHistory.service_id)
.select_from(NotificationHistory)
.filter(
.where(
NotificationHistory.created_at >= start_date,
NotificationHistory.created_at <= end_date,
NotificationHistory.notification_type.in_(
@@ -568,7 +568,7 @@ def fetch_email_usage_for_organization(organization_id, start_date, end_date):
FactBilling,
FactBilling.service_id == Service.id,
)
.filter(
.where(
FactBilling.local_date >= start_date,
FactBilling.local_date <= end_date,
FactBilling.notification_type == NotificationType.EMAIL,
@@ -586,12 +586,12 @@ def fetch_email_usage_for_organization(organization_id, start_date, end_date):
def fetch_sms_billing_for_organization(organization_id, financial_year):
# ASSUMPTION: AnnualBilling has been populated for year.
ft_billing_subquery = query_organization_sms_usage_for_year(
ft_billing_substmt = query_organization_sms_usage_for_year(
organization_id, financial_year
).subquery()
sms_billable_units = func.sum(
func.coalesce(ft_billing_subquery.c.chargeable_units, 0)
func.coalesce(ft_billing_substmt.c.chargeable_units, 0)
)
# subtract sms_billable_units units accrued since report's start date to get up-to-date
@@ -600,8 +600,8 @@ def fetch_sms_billing_for_organization(organization_id, financial_year):
AnnualBilling.free_sms_fragment_limit - sms_billable_units, 0
)
chargeable_sms = func.sum(ft_billing_subquery.c.charged_units)
sms_cost = func.sum(ft_billing_subquery.c.cost)
chargeable_sms = func.sum(ft_billing_substmt.c.charged_units)
sms_cost = func.sum(ft_billing_substmt.c.cost)
query = (
select(
@@ -622,8 +622,8 @@ def fetch_sms_billing_for_organization(organization_id, financial_year):
AnnualBilling.financial_year_start == financial_year,
),
)
.outerjoin(ft_billing_subquery, Service.id == ft_billing_subquery.c.service_id)
.filter(
.outerjoin(ft_billing_substmt, Service.id == ft_billing_substmt.c.service_id)
.where(
Service.organization_id == organization_id, Service.restricted.is_(False)
)
.group_by(Service.id, Service.name, AnnualBilling.free_sms_fragment_limit)
@@ -688,7 +688,7 @@ def query_organization_sms_usage_for_year(organization_id, year):
FactBilling.notification_type == NotificationType.SMS,
),
)
.filter(
.where(
Service.organization_id == organization_id,
AnnualBilling.financial_year_start == year,
)
@@ -812,9 +812,7 @@ def fetch_daily_volumes_for_platform(start_date, end_date):
)
).label("email_totals"),
)
.filter(
FactBilling.local_date >= start_date, FactBilling.local_date <= end_date
)
.where(FactBilling.local_date >= start_date, FactBilling.local_date <= end_date)
.group_by(FactBilling.local_date, FactBilling.notification_type)
.subquery()
)
@@ -857,7 +855,7 @@ def fetch_daily_sms_provider_volumes_for_platform(start_date, end_date):
).label("sms_cost"),
)
.select_from(FactBilling)
.filter(
.where(
FactBilling.notification_type == NotificationType.SMS,
FactBilling.local_date >= start_date,
FactBilling.local_date <= end_date,
@@ -912,9 +910,7 @@ def fetch_volumes_by_service(start_date, end_date):
).label("email_totals"),
)
.select_from(FactBilling)
.filter(
FactBilling.local_date >= start_date, FactBilling.local_date <= end_date
)
.where(FactBilling.local_date >= start_date, FactBilling.local_date <= end_date)
.group_by(
FactBilling.local_date,
FactBilling.service_id,
@@ -930,7 +926,7 @@ def fetch_volumes_by_service(start_date, end_date):
AnnualBilling.free_sms_fragment_limit,
)
.select_from(AnnualBilling)
.filter(AnnualBilling.financial_year_start <= year_end_date)
.where(AnnualBilling.financial_year_start <= year_end_date)
.group_by(AnnualBilling.service_id, AnnualBilling.free_sms_fragment_limit)
.subquery()
)
@@ -957,7 +953,7 @@ def fetch_volumes_by_service(start_date, end_date):
.outerjoin( # include services without volume
volume_stats, Service.id == volume_stats.c.service_id
)
.filter(
.where(
Service.restricted.is_(False),
Service.count_as_live.is_(True),
Service.active.is_(True),

58
app/dao/fact_notification_status_dao.py
View File

@@ -33,7 +33,7 @@ def update_fact_notification_status(process_day, notification_type, service_id):
end_date = get_midnight_in_utc(process_day + timedelta(days=1))
# delete any existing rows in case some no longer exist e.g. if all messages are sent
stmt = delete(FactNotificationStatus).filter(
stmt = delete(FactNotificationStatus).where(
FactNotificationStatus.local_date == process_day,
FactNotificationStatus.notification_type == notification_type,
FactNotificationStatus.service_id == service_id,
@@ -55,7 +55,7 @@ def update_fact_notification_status(process_day, notification_type, service_id):
func.count().label("notification_count"),
)
.select_from(NotificationAllTimeView)
.filter(
.where(
NotificationAllTimeView.created_at >= start_date,
NotificationAllTimeView.created_at < end_date,
NotificationAllTimeView.notification_type == notification_type,
@@ -97,7 +97,7 @@ def fetch_notification_status_for_service_by_month(start_date, end_date, service
func.count(NotificationAllTimeView.id).label("count"),
)
.select_from(NotificationAllTimeView)
.filter(
.where(
NotificationAllTimeView.service_id == service_id,
NotificationAllTimeView.created_at >= start_date,
NotificationAllTimeView.created_at < end_date,
@@ -122,7 +122,7 @@ def fetch_notification_status_for_service_for_day(fetch_day, service_id):
func.count().label("count"),
)
.select_from(Notification)
.filter(
.where(
Notification.created_at >= get_midnight_in_utc(fetch_day),
Notification.created_at
< get_midnight_in_utc(fetch_day + timedelta(days=1)),
@@ -191,7 +191,7 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(
all_stats_alias = aliased(all_stats_union, name="all_stats")
# Final query with optional template joins
query = select(
stmt = select(
*(
[
TemplateFolder.name.label("folder"),
@@ -214,8 +214,8 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(
)
if by_template:
query = (
query.join(Template, all_stats_alias.c.template_id == Template.id)
stmt = (
stmt.join(Template, all_stats_alias.c.template_id == Template.id)
.join(User, Template.created_by_id == User.id)
.outerjoin(
template_folder_map, Template.id == template_folder_map.c.template_id
@@ -227,7 +227,7 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(
)
# Group by all necessary fields except date_used
query = query.group_by(
stmt = stmt.group_by(
*(
[
TemplateFolder.name,
@@ -245,7 +245,7 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(
)
# Execute the query using Flask-SQLAlchemy's session
result = db.session.execute(query)
result = db.session.execute(stmt)
return result.mappings().all()
@@ -260,7 +260,7 @@ def fetch_notification_status_totals_for_all_services(start_date, end_date):
func.sum(FactNotificationStatus.notification_count).label("count"),
)
.select_from(FactNotificationStatus)
.filter(
.where(
FactNotificationStatus.local_date >= start_date,
FactNotificationStatus.local_date <= end_date,
)
@@ -279,7 +279,7 @@ def fetch_notification_status_totals_for_all_services(start_date, end_date):
Notification.key_type.cast(db.Text),
func.count().label("count"),
)
.filter(Notification.created_at >= today)
.where(Notification.created_at >= today)
.group_by(
Notification.notification_type,
Notification.status,
@@ -313,7 +313,7 @@ def fetch_notification_statuses_for_job(job_id):
func.sum(FactNotificationStatus.notification_count).label("count"),
)
.select_from(FactNotificationStatus)
.filter(
.where(
FactNotificationStatus.job_id == job_id,
)
.group_by(FactNotificationStatus.notification_status)
@@ -338,7 +338,7 @@ def fetch_stats_for_all_services_by_date_range(
func.sum(FactNotificationStatus.notification_count).label("count"),
)
.select_from(FactNotificationStatus)
.filter(
.where(
FactNotificationStatus.local_date >= start_date,
FactNotificationStatus.local_date <= end_date,
FactNotificationStatus.service_id == Service.id,
@@ -357,11 +357,11 @@ def fetch_stats_for_all_services_by_date_range(
)
)
if not include_from_test_key:
stats = stats.filter(FactNotificationStatus.key_type != KeyType.TEST)
stats = stats.where(FactNotificationStatus.key_type != KeyType.TEST)
if start_date <= utc_now().date() <= end_date:
today = get_midnight_in_utc(utc_now())
subquery = (
substmt = (
select(
Notification.notification_type.label("notification_type"),
Notification.status.label("status"),
@@ -369,7 +369,7 @@ def fetch_stats_for_all_services_by_date_range(
func.count(Notification.id).label("count"),
)
.select_from(Notification)
.filter(Notification.created_at >= today)
.where(Notification.created_at >= today)
.group_by(
Notification.notification_type,
Notification.status,
@@ -377,8 +377,8 @@ def fetch_stats_for_all_services_by_date_range(
)
)
if not include_from_test_key:
subquery = subquery.filter(Notification.key_type != KeyType.TEST)
subquery = subquery.subquery()
substmt = substmt.where(Notification.key_type != KeyType.TEST)
substmt = substmt.subquery()
stats_for_today = select(
Service.id.label("service_id"),
@@ -386,10 +386,10 @@ def fetch_stats_for_all_services_by_date_range(
Service.restricted.label("restricted"),
Service.active.label("active"),
Service.created_at.label("created_at"),
subquery.c.notification_type.cast(db.Text).label("notification_type"),
subquery.c.status.cast(db.Text).label("status"),
subquery.c.count.label("count"),
).outerjoin(subquery, subquery.c.service_id == Service.id)
substmt.c.notification_type.cast(db.Text).label("notification_type"),
substmt.c.status.cast(db.Text).label("status"),
substmt.c.count.label("count"),
).outerjoin(substmt, substmt.c.service_id == Service.id)
all_stats_table = stats.union_all(stats_for_today).subquery()
query = (
@@ -435,7 +435,7 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
func.sum(FactNotificationStatus.notification_count).label("count"),
)
.join(Template, FactNotificationStatus.template_id == Template.id)
.filter(
.where(
FactNotificationStatus.service_id == service_id,
FactNotificationStatus.local_date >= start_date,
FactNotificationStatus.local_date <= end_date,
@@ -473,7 +473,7 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
Template,
Notification.template_id == Template.id,
)
.filter(
.where(
Notification.created_at >= today,
Notification.service_id == service_id,
Notification.key_type != KeyType.TEST,
@@ -515,7 +515,7 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
def get_total_notifications_for_date_range(start_date, end_date):
query = (
stmt = (
select(
FactNotificationStatus.local_date.label("local_date"),
func.sum(
@@ -539,18 +539,18 @@ def get_total_notifications_for_date_range(start_date, end_date):
)
).label("sms"),
)
.filter(
.where(
FactNotificationStatus.key_type != KeyType.TEST,
)
.group_by(FactNotificationStatus.local_date)
.order_by(FactNotificationStatus.local_date)
)
if start_date and end_date:
query = query.filter(
stmt = stmt.where(
FactNotificationStatus.local_date >= start_date,
FactNotificationStatus.local_date <= end_date,
)
return db.session.execute(query).all()
return db.session.execute(stmt).all()
def fetch_monthly_notification_statuses_per_service(start_date, end_date):
@@ -629,7 +629,7 @@ def fetch_monthly_notification_statuses_per_service(start_date, end_date):
).label("count_sent"),
)
.join(Service, FactNotificationStatus.service_id == Service.id)
.filter(
.where(
FactNotificationStatus.notification_status != NotificationStatus.CREATED,
Service.active.is_(True),
FactNotificationStatus.key_type != KeyType.TEST,

7
app/dao/fact_processing_time_dao.py
View File

@@ -1,3 +1,4 @@
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.sql.expression import case
@@ -34,7 +35,7 @@ def insert_update_processing_time(processing_time):
def get_processing_time_percentage_for_date_range(start_date, end_date):
query = (
db.session.query(
select(
FactProcessingTime.local_date.cast(db.Text).label("date"),
FactProcessingTime.messages_total,
FactProcessingTime.messages_within_10_secs,
@@ -52,11 +53,11 @@ def get_processing_time_percentage_for_date_range(start_date, end_date):
(FactProcessingTime.messages_total == 0, 100.0),
).label("percentage"),
)
.filter(
.where(
FactProcessingTime.local_date >= start_date,
FactProcessingTime.local_date <= end_date,
)
.order_by(FactProcessingTime.local_date)
)
return query.all()
return db.session.execute(query).all()

8
app/dao/inbound_numbers_dao.py
View File

@@ -11,19 +11,19 @@ def dao_get_inbound_numbers():
def dao_get_available_inbound_numbers():
stmt = select(InboundNumber).filter(
stmt = select(InboundNumber).where(
InboundNumber.active, InboundNumber.service_id.is_(None)
)
return db.session.execute(stmt).scalars().all()
def dao_get_inbound_number_for_service(service_id):
stmt = select(InboundNumber).filter(InboundNumber.service_id == service_id)
stmt = select(InboundNumber).where(InboundNumber.service_id == service_id)
return db.session.execute(stmt).scalars().first()
def dao_get_inbound_number(inbound_number_id):
stmt = select(InboundNumber).filter(InboundNumber.id == inbound_number_id)
stmt = select(InboundNumber).where(InboundNumber.id == inbound_number_id)
return db.session.execute(stmt).scalars().first()
@@ -35,7 +35,7 @@ def dao_set_inbound_number_to_service(service_id, inbound_number):
@autocommit
def dao_set_inbound_number_active_flag(service_id, active):
stmt = select(InboundNumber).filter(InboundNumber.service_id == service_id)
stmt = select(InboundNumber).where(InboundNumber.service_id == service_id)
inbound_number = db.session.execute(stmt).scalars().first()
inbound_number.active = active

42
app/dao/inbound_sms_dao.py
View File

@@ -20,15 +20,15 @@ def dao_get_inbound_sms_for_service(
):
q = (
select(InboundSms)
.filter(InboundSms.service_id == service_id)
.where(InboundSms.service_id == service_id)
.order_by(InboundSms.created_at.desc())
)
if limit_days is not None:
start_date = midnight_n_days_ago(limit_days)
q = q.filter(InboundSms.created_at >= start_date)
q = q.where(InboundSms.created_at >= start_date)
if user_number:
q = q.filter(InboundSms.user_number == user_number)
q = q.where(InboundSms.user_number == user_number)
if limit:
q = q.limit(limit)
@@ -47,22 +47,32 @@ def dao_get_paginated_inbound_sms_for_service_for_public_api(
if older_than:
older_than_created_at = (
db.session.query(InboundSms.created_at)
.filter(InboundSms.id == older_than)
.where(InboundSms.id == older_than)
.scalar_subquery()
)
filters.append(InboundSms.created_at < older_than_created_at)
page = 1 # ?
offset = (page - 1) * page_size
# As part of the move to sqlalchemy 2.0, we do this manual pagination
query = db.session.query(InboundSms).filter(*filters)
paginated_items = query.order_by(desc(InboundSms.created_at)).limit(page_size).all()
return paginated_items
stmt = (
select(InboundSms)
.where(*filters)
.order_by(desc(InboundSms.created_at))
.limit(page_size)
.offset(offset)
)
paginated_items = db.session.execute(stmt).scalars().all()
total_items = db.session.execute(select(func.count()).where(*filters)).scalar() or 0
pagination = Pagination(paginated_items, page, page_size, total_items)
return pagination
def dao_count_inbound_sms_for_service(service_id, limit_days):
stmt = (
select(func.count())
.select_from(InboundSms)
.filter(
.where(
InboundSms.service_id == service_id,
InboundSms.created_at >= midnight_n_days_ago(limit_days),
)
@@ -74,7 +84,7 @@ def dao_count_inbound_sms_for_service(service_id, limit_days):
def _insert_inbound_sms_history(subquery, query_limit=10000):
offset = 0
subquery_select = select(subquery)
inbound_sms_query = select(
inbound_sms_stmt = select(
InboundSms.id,
InboundSms.created_at,
InboundSms.service_id,
@@ -84,13 +94,13 @@ def _insert_inbound_sms_history(subquery, query_limit=10000):
InboundSms.provider,
).where(InboundSms.id.in_(subquery_select))
count_query = select(func.count()).select_from(inbound_sms_query.subquery())
count_query = select(func.count()).select_from(inbound_sms_stmt.subquery())
inbound_sms_count = db.session.execute(count_query).scalar() or 0
while offset < inbound_sms_count:
statement = insert(InboundSmsHistory).from_select(
InboundSmsHistory.__table__.c,
inbound_sms_query.limit(query_limit).offset(offset),
inbound_sms_stmt.limit(query_limit).offset(offset),
)
statement = statement.on_conflict_do_nothing(
@@ -107,7 +117,7 @@ def _delete_inbound_sms(datetime_to_delete_from, query_filter):
subquery = (
select(InboundSms.id)
.filter(InboundSms.created_at < datetime_to_delete_from, *query_filter)
.where(InboundSms.created_at < datetime_to_delete_from, *query_filter)
.limit(query_limit)
.subquery()
)
@@ -118,7 +128,7 @@ def _delete_inbound_sms(datetime_to_delete_from, query_filter):
while number_deleted > 0:
_insert_inbound_sms_history(subquery, query_limit=query_limit)
stmt = delete(InboundSms).filter(InboundSms.id.in_(subquery))
stmt = delete(InboundSms).where(InboundSms.id.in_(subquery))
number_deleted = db.session.execute(stmt).rowcount
db.session.commit()
deleted += number_deleted
@@ -135,7 +145,7 @@ def delete_inbound_sms_older_than_retention():
stmt = (
select(ServiceDataRetention)
.join(ServiceDataRetention.service)
.filter(ServiceDataRetention.notification_type == NotificationType.SMS)
.where(ServiceDataRetention.notification_type == NotificationType.SMS)
)
flexible_data_retention = db.session.execute(stmt).scalars().all()
@@ -170,7 +180,9 @@ def delete_inbound_sms_older_than_retention():
def dao_get_inbound_sms_by_id(service_id, inbound_id):
stmt = select(InboundSms).filter_by(id=inbound_id, service_id=service_id)
stmt = select(InboundSms).where(
InboundSms.id == inbound_id, InboundSms.service_id == service_id
)
return db.session.execute(stmt).scalars().one()

39
app/dao/invited_org_user_dao.py
View File

@@ -1,5 +1,7 @@
from datetime import timedelta
from sqlalchemy import select
from app import db
from app.models import InvitedOrganizationUser
from app.utils import utc_now
@@ -11,25 +13,46 @@ def save_invited_org_user(invited_org_user):
def get_invited_org_user(organization_id, invited_org_user_id):
return InvitedOrganizationUser.query.filter_by(
organization_id=organization_id, id=invited_org_user_id
).one()
return (
db.session.execute(
select(InvitedOrganizationUser).where(
InvitedOrganizationUser.organization_id == organization_id,
InvitedOrganizationUser.id == invited_org_user_id,
)
)
.scalars()
.one()
)
def get_invited_org_user_by_id(invited_org_user_id):
return InvitedOrganizationUser.query.filter_by(id=invited_org_user_id).one()
return (
db.session.execute(
select(InvitedOrganizationUser).where(
InvitedOrganizationUser.id == invited_org_user_id
)
)
.scalars()
.one()
)
def get_invited_org_users_for_organization(organization_id):
return InvitedOrganizationUser.query.filter_by(
organization_id=organization_id
).all()
return (
db.session.execute(
select(InvitedOrganizationUser).where(
InvitedOrganizationUser.organization_id == organization_id
)
)
.scalars()
.all()
)
def delete_org_invitations_created_more_than_two_days_ago():
deleted = (
db.session.query(InvitedOrganizationUser)
.filter(InvitedOrganizationUser.created_at <= utc_now() - timedelta(days=2))
.where(InvitedOrganizationUser.created_at <= utc_now() - timedelta(days=2))
.delete()
)
db.session.commit()

2
app/dao/invited_user_dao.py
View File

@@ -50,7 +50,7 @@ def get_invited_users_for_service(service_id):
def expire_invitations_created_more_than_two_days_ago():
expired = (
db.session.query(InvitedUser)
.filter(
.where(
InvitedUser.created_at <= utc_now() - timedelta(days=2),
InvitedUser.status.in_((InvitedUserStatus.PENDING,)),
)

39
app/dao/jobs_dao.py
View File

@@ -3,7 +3,7 @@ import uuid
from datetime import timedelta
from flask import current_app
from sqlalchemy import and_, asc, desc, func, select
from sqlalchemy import and_, asc, desc, func, select, update
from app import db
from app.dao.pagination import Pagination
@@ -21,7 +21,7 @@ from app.utils import midnight_n_days_ago, utc_now
def dao_get_notification_outcomes_for_job(service_id, job_id):
stmt = (
select(func.count(Notification.status).label("count"), Notification.status)
.filter(Notification.service_id == service_id, Notification.job_id == job_id)
.where(Notification.service_id == service_id, Notification.job_id == job_id)
.group_by(Notification.status)
)
notification_statuses = db.session.execute(stmt).all()
@@ -30,7 +30,7 @@ def dao_get_notification_outcomes_for_job(service_id, job_id):
stmt = select(
FactNotificationStatus.notification_count.label("count"),
FactNotificationStatus.notification_status.label("status"),
).filter(
).where(
FactNotificationStatus.service_id == service_id,
FactNotificationStatus.job_id == job_id,
)
@@ -39,13 +39,14 @@ def dao_get_notification_outcomes_for_job(service_id, job_id):
def dao_get_job_by_service_id_and_job_id(service_id, job_id):
stmt = select(Job).filter_by(service_id=service_id, id=job_id)
stmt = select(Job).where(Job.service_id == service_id, Job.id == job_id)
return db.session.execute(stmt).scalars().one()
def dao_get_unfinished_jobs():
stmt = select(Job).filter(Job.processing_finished.is_(None))
return db.session.execute(stmt).all()
return db.session.execute(stmt).scalars().all()
def dao_get_jobs_by_service_id(
@@ -67,13 +68,13 @@ def dao_get_jobs_by_service_id(
query_filter.append(Job.job_status.in_(statuses))
total_items = db.session.execute(
select(func.count()).select_from(Job).filter(*query_filter)
select(func.count()).select_from(Job).where(*query_filter)
).scalar_one()
offset = (page - 1) * page_size
stmt = (
select(Job)
.filter(*query_filter)
.where(*query_filter)
.order_by(Job.processing_started.desc(), Job.created_at.desc())
.limit(page_size)
.offset(offset)
@@ -89,7 +90,7 @@ def dao_get_scheduled_job_stats(
stmt = select(
func.count(Job.id),
func.min(Job.scheduled_for),
).filter(
).where(
Job.service_id == service_id,
Job.job_status == JobStatus.SCHEDULED,
)
@@ -97,7 +98,7 @@ def dao_get_scheduled_job_stats(
def dao_get_job_by_id(job_id):
stmt = select(Job).filter_by(id=job_id)
stmt = select(Job).where(Job.id == job_id)
return db.session.execute(stmt).scalars().one()
@@ -117,7 +118,7 @@ def dao_set_scheduled_jobs_to_pending():
"""
stmt = (
select(Job)
.filter(
.where(
Job.job_status == JobStatus.SCHEDULED,
Job.scheduled_for < utc_now(),
)
@@ -136,7 +137,7 @@ def dao_set_scheduled_jobs_to_pending():
def dao_get_future_scheduled_job_by_id_and_service_id(job_id, service_id):
stmt = select(Job).filter(
stmt = select(Job).where(
Job.service_id == service_id,
Job.id == job_id,
Job.job_status == JobStatus.SCHEDULED,
@@ -176,8 +177,14 @@ def dao_update_job(job):
db.session.commit()
def dao_update_job_status_to_error(job):
stmt = update(Job).where(Job.id == job.id).values(job_status=JobStatus.ERROR)
db.session.execute(stmt)
db.session.commit()
def dao_get_jobs_older_than_data_retention(notification_types):
stmt = select(ServiceDataRetention).filter(
stmt = select(ServiceDataRetention).where(
ServiceDataRetention.notification_type.in_(notification_types)
)
flexible_data_retention = db.session.execute(stmt).scalars().all()
@@ -188,7 +195,7 @@ def dao_get_jobs_older_than_data_retention(notification_types):
stmt = (
select(Job)
.join(Template)
.filter(
.where(
func.coalesce(Job.scheduled_for, Job.created_at) < end_date,
Job.archived == False, # noqa
Template.template_type == f.notification_type,
@@ -209,7 +216,7 @@ def dao_get_jobs_older_than_data_retention(notification_types):
stmt = (
select(Job)
.join(Template)
.filter(
.where(
func.coalesce(Job.scheduled_for, Job.created_at) < end_date,
Job.archived == False, # noqa
Template.template_type == notification_type,
@@ -229,7 +236,7 @@ def find_jobs_with_missing_rows():
yesterday = utc_now() - timedelta(days=1)
jobs_with_rows_missing = (
select(Job)
.filter(
.where(
Job.job_status == JobStatus.FINISHED,
Job.processing_finished < ten_minutes_ago,
Job.processing_finished > yesterday,
@@ -258,6 +265,6 @@ def find_missing_row_for_job(job_id, job_size):
Notification.job_id == job_id,
),
)
.filter(Notification.job_row_number == None) # noqa
.where(Notification.job_row_number == None) # noqa
)
return db.session.execute(query).all()

172
app/dao/notifications_dao.py
View File

@@ -1,5 +1,6 @@
import json
from datetime import timedelta
import os
from datetime import datetime, timedelta
from time import time
from flask import current_app
@@ -24,6 +25,7 @@ from werkzeug.datastructures import MultiDict
from app import create_uuid, db
from app.dao.dao_utils import autocommit
from app.dao.inbound_sms_dao import Pagination
from app.enums import KeyType, NotificationStatus, NotificationType
from app.models import FactNotificationStatus, Notification, NotificationHistory
from app.utils import (
@@ -43,7 +45,7 @@ from notifications_utils.recipients import (
def dao_get_last_date_template_was_used(template_id, service_id):
last_date_from_notifications = (
db.session.query(functions.max(Notification.created_at))
.filter(
.where(
Notification.service_id == service_id,
Notification.template_id == template_id,
Notification.key_type != KeyType.TEST,
@@ -56,7 +58,7 @@ def dao_get_last_date_template_was_used(template_id, service_id):
last_date = (
db.session.query(functions.max(FactNotificationStatus.local_date))
.filter(
.where(
FactNotificationStatus.template_id == template_id,
FactNotificationStatus.key_type != KeyType.TEST,
)
@@ -95,6 +97,32 @@ def dao_create_notification(notification):
# notify-api-1454 insert only if it doesn't exist
if not dao_notification_exists(notification.id):
db.session.add(notification)
# There have been issues with invites expiring.
# Ensure the created at value is set and debug.
if notification.notification_type == "email":
orig_time = notification.created_at
now_time = utc_now()
try:
diff_time = now_time - orig_time
except TypeError:
try:
orig_time = datetime.strptime(orig_time, "%Y-%m-%dT%H:%M:%S.%fZ")
except ValueError:
orig_time = datetime.strptime(orig_time, "%Y-%m-%d")
diff_time = now_time - orig_time
current_app.logger.error(
f"dao_create_notification orig created at: {orig_time} and now created at: {now_time}"
)
if diff_time.total_seconds() > 300:
current_app.logger.error(
"Something is wrong with notification.created_at in email!"
)
if os.getenv("NOTIFY_ENVIRONMENT") not in ["test"]:
notification.created_at = now_time
dao_update_notification(notification)
current_app.logger.error(
f"Email notification created_at reset to {notification.created_at}"
)
def country_records_delivery(phone_prefix):
@@ -143,9 +171,7 @@ def update_notification_status_by_id(
notification_id, status, sent_by=None, provider_response=None, carrier=None
):
stmt = (
select(Notification)
.with_for_update()
.filter(Notification.id == notification_id)
select(Notification).with_for_update().where(Notification.id == notification_id)
)
notification = db.session.execute(stmt).scalars().first()
@@ -190,7 +216,7 @@ def update_notification_status_by_id(
@autocommit
def update_notification_status_by_reference(reference, status):
# this is used to update emails
stmt = select(Notification).filter(Notification.reference == reference)
stmt = select(Notification).where(Notification.reference == reference)
notification = db.session.execute(stmt).scalars().first()
if not notification:
@@ -226,40 +252,59 @@ def get_notifications_for_job(
if page_size is None:
page_size = current_app.config["PAGE_SIZE"]
query = Notification.query.filter_by(service_id=service_id, job_id=job_id)
query = _filter_query(query, filter_dict)
return query.order_by(asc(Notification.job_row_number)).paginate(
page=page, per_page=page_size
stmt = select(Notification).where(
Notification.service_id == service_id, Notification.job_id == job_id
)
stmt = _filter_query(stmt, filter_dict)
stmt = stmt.order_by(asc(Notification.job_row_number))
results = db.session.execute(stmt).scalars().all()
page_size = current_app.config["PAGE_SIZE"]
offset = (page - 1) * page_size
paginated_results = results[offset : offset + page_size]
pagination = Pagination(paginated_results, page, page_size, len(results))
return pagination
def dao_get_notification_count_for_job_id(*, job_id):
stmt = select(func.count(Notification.id)).filter_by(job_id=job_id)
stmt = select(func.count(Notification.id)).where(Notification.job_id == job_id)
return db.session.execute(stmt).scalar()
def dao_get_notification_count_for_service(*, service_id):
stmt = select(func.count(Notification.id)).filter_by(service_id=service_id)
stmt = select(func.count(Notification.id)).where(
Notification.service_id == service_id
)
return db.session.execute(stmt).scalar()
def dao_get_failed_notification_count():
stmt = select(func.count(Notification.id)).filter_by(
status=NotificationStatus.FAILED
stmt = select(func.count(Notification.id)).where(
Notification.status == NotificationStatus.FAILED
)
return db.session.execute(stmt).scalar()
def get_notification_with_personalisation(service_id, notification_id, key_type):
filter_dict = {"service_id": service_id, "id": notification_id}
if key_type:
filter_dict["key_type"] = key_type
stmt = (
select(Notification)
.filter_by(**filter_dict)
.where(
Notification.service_id == service_id, Notification.id == notification_id
)
.options(joinedload(Notification.template))
)
if key_type:
stmt = (
select(Notification)
.where(
Notification.service_id == service_id,
Notification.id == notification_id,
Notification.key_type == key_type,
)
.options(joinedload(Notification.template))
)
return db.session.execute(stmt).scalars().one()
@@ -269,7 +314,7 @@ def get_notification_by_id(notification_id, service_id=None, _raise=False):
if service_id:
filters.append(Notification.service_id == service_id)
stmt = select(Notification).filter(*filters)
stmt = select(Notification).where(*filters)
return (
db.session.execute(stmt).scalars().one()
@@ -305,7 +350,7 @@ def get_notifications_for_service(
if older_than is not None:
older_than_created_at = (
db.session.query(Notification.created_at)
.filter(Notification.id == older_than)
.where(Notification.id == older_than)
.as_scalar()
)
filters.append(Notification.created_at < older_than_created_at)
@@ -324,22 +369,22 @@ def get_notifications_for_service(
if client_reference is not None:
filters.append(Notification.client_reference == client_reference)
query = Notification.query.filter(*filters)
query = _filter_query(query, filter_dict)
stmt = select(Notification).where(*filters)
stmt = _filter_query(stmt, filter_dict)
if personalisation:
query = query.options(joinedload(Notification.template))
stmt = stmt.options(joinedload(Notification.template))
return query.order_by(desc(Notification.created_at)).paginate(
page=page,
per_page=page_size,
count=count_pages,
error_out=error_out,
)
stmt = stmt.order_by(desc(Notification.created_at))
results = db.session.execute(stmt).scalars().all()
offset = (page - 1) * page_size
paginated_results = results[offset : offset + page_size]
pagination = Pagination(paginated_results, page, page_size, len(results))
return pagination
def _filter_query(query, filter_dict=None):
def _filter_query(stmt, filter_dict=None):
if filter_dict is None:
return query
return stmt
multidict = MultiDict(filter_dict)
@@ -347,14 +392,14 @@ def _filter_query(query, filter_dict=None):
statuses = multidict.getlist("status")
if statuses:
query = query.filter(Notification.status.in_(statuses))
stmt = stmt.where(Notification.status.in_(statuses))
# filter by template
template_types = multidict.getlist("template_type")
if template_types:
query = query.filter(Notification.notification_type.in_(template_types))
stmt = stmt.where(Notification.notification_type.in_(template_types))
return query
return stmt
def sanitize_successful_notification_by_id(notification_id, carrier, provider_response):
@@ -455,7 +500,7 @@ def move_notifications_to_notification_history(
deleted += delete_count_per_call
# Deleting test Notifications, test notifications are not persisted to NotificationHistory
stmt = delete(Notification).filter(
stmt = delete(Notification).where(
Notification.notification_type == notification_type,
Notification.service_id == service_id,
Notification.created_at < timestamp_to_delete_backwards_from,
@@ -469,7 +514,7 @@ def move_notifications_to_notification_history(
@autocommit
def dao_delete_notifications_by_id(notification_id):
db.session.query(Notification).filter(Notification.id == notification_id).delete(
db.session.query(Notification).where(Notification.id == notification_id).delete(
synchronize_session="fetch"
)
@@ -485,7 +530,7 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
stmt = (
select(Notification)
.filter(
.where(
Notification.created_at < cutoff_time,
Notification.status.in_(current_statuses),
Notification.notification_type.in_(
@@ -498,7 +543,7 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
stmt = (
update(Notification)
.filter(Notification.id.in_([n.id for n in notifications]))
.where(Notification.id.in_([n.id for n in notifications]))
.values({"status": new_status, "updated_at": updated_at})
)
db.session.execute(stmt)
@@ -511,7 +556,7 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
def dao_update_notifications_by_reference(references, update_dict):
stmt = (
update(Notification)
.filter(Notification.reference.in_(references))
.where(Notification.reference.in_(references))
.values(update_dict)
)
result = db.session.execute(stmt)
@@ -521,7 +566,7 @@ def dao_update_notifications_by_reference(references, update_dict):
if updated_count != len(references):
stmt = (
update(NotificationHistory)
.filter(NotificationHistory.reference.in_(references))
.where(NotificationHistory.reference.in_(references))
.values(update_dict)
)
result = db.session.execute(stmt)
@@ -584,7 +629,7 @@ def dao_get_notifications_by_recipient_or_reference(
results = (
db.session.query(Notification)
.filter(*filters)
.where(*filters)
.order_by(desc(Notification.created_at))
.paginate(page=page, per_page=page_size, count=False, error_out=error_out)
)
@@ -592,7 +637,7 @@ def dao_get_notifications_by_recipient_or_reference(
def dao_get_notification_by_reference(reference):
stmt = select(Notification).filter(Notification.reference == reference)
stmt = select(Notification).where(Notification.reference == reference)
return db.session.execute(stmt).scalars().one()
@@ -600,10 +645,10 @@ 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
stmt = select(Notification).filter(Notification.reference == reference)
stmt = select(Notification).where(Notification.reference == reference)
return db.session.execute(stmt).scalars().one()
except NoResultFound:
stmt = select(NotificationHistory).filter(
stmt = select(NotificationHistory).where(
NotificationHistory.reference == reference
)
return db.session.execute(stmt).scalars().one()
@@ -646,7 +691,7 @@ def dao_get_notifications_processing_time_stats(start_date, end_date):
def dao_get_last_notification_added_for_job_id(job_id):
stmt = (
select(Notification)
.filter(Notification.job_id == job_id)
.where(Notification.job_id == job_id)
.order_by(Notification.job_row_number.desc())
)
last_notification_added = db.session.execute(stmt).scalars().first()
@@ -657,7 +702,7 @@ 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)
stmt = select(Notification).filter(
stmt = select(Notification).where(
Notification.created_at <= older_than_date,
Notification.notification_type == notification_type,
Notification.status == NotificationStatus.CREATED,
@@ -689,7 +734,7 @@ def get_service_ids_with_notifications_before(notification_type, timestamp):
return {
row.service_id
for row in db.session.query(Notification.service_id)
.filter(
.where(
Notification.notification_type == notification_type,
Notification.created_at < timestamp,
)
@@ -703,7 +748,7 @@ def get_service_ids_with_notifications_on_date(notification_type, date):
notification_table_query = db.session.query(
Notification.service_id.label("service_id")
).filter(
).where(
Notification.notification_type == notification_type,
# using >= + < is much more efficient than date(created_at)
Notification.created_at >= start_date,
@@ -714,7 +759,7 @@ def get_service_ids_with_notifications_on_date(notification_type, date):
# provided the task to populate it has run before they were archived.
ft_status_table_query = db.session.query(
FactNotificationStatus.service_id.label("service_id")
).filter(
).where(
FactNotificationStatus.notification_type == notification_type,
FactNotificationStatus.local_date == date,
)
@@ -780,3 +825,30 @@ def dao_update_delivery_receipts(receipts, delivered):
f"#loadtestperformance batch update query time: \
updated {len(receipts)} notification in {elapsed_time} ms"
)
def dao_close_out_delivery_receipts():
THREE_DAYS_AGO = utc_now() - timedelta(minutes=3)
stmt = (
update(Notification)
.where(
Notification.status == NotificationStatus.PENDING,
Notification.sent_at < THREE_DAYS_AGO,
)
.values(status=NotificationStatus.FAILED, provider_response="Technical Failure")
)
result = db.session.execute(stmt)
db.session.commit()
if result:
current_app.logger.info(
f"Marked {result.rowcount} notifications as technical failures"
)
def dao_batch_insert_notifications(batch):
db.session.bulk_save_objects(batch)
db.session.commit()
current_app.logger.info(f"Batch inserted notifications: {len(batch)}")
return len(batch)

24
app/dao/organization_dao.py
View File

@@ -17,7 +17,7 @@ def dao_count_organizations_with_live_services():
stmt = (
select(func.count(func.distinct(Organization.id)))
.join(Organization.services)
.filter(
.where(
Service.active.is_(True),
Service.restricted.is_(False),
Service.count_as_live.is_(True),
@@ -27,17 +27,19 @@ def dao_count_organizations_with_live_services():
def dao_get_organization_services(organization_id):
stmt = select(Organization).filter_by(id=organization_id)
stmt = select(Organization).where(Organization.id == organization_id)
return db.session.execute(stmt).scalars().one().services
def dao_get_organization_live_services(organization_id):
stmt = select(Service).filter_by(organization_id=organization_id, restricted=False)
stmt = select(Service).where(
Service.organization_id == organization_id, Service.restricted == False # noqa
)
return db.session.execute(stmt).scalars().all()
def dao_get_organization_by_id(organization_id):
stmt = select(Organization).filter_by(id=organization_id)
stmt = select(Organization).where(Organization.id == organization_id)
return db.session.execute(stmt).scalars().one()
@@ -49,14 +51,16 @@ def dao_get_organization_by_email_address(email_address):
if email_address.endswith(
"@{}".format(domain.domain)
) or email_address.endswith(".{}".format(domain.domain)):
stmt = select(Organization).filter_by(id=domain.organization_id)
stmt = select(Organization).where(Organization.id == domain.organization_id)
return db.session.execute(stmt).scalars().one()
return None
def dao_get_organization_by_service_id(service_id):
stmt = select(Organization).join(Organization.services).filter_by(id=service_id)
stmt = (
select(Organization).join(Organization.services).where(Service.id == service_id)
)
return db.session.execute(stmt).scalars().first()
@@ -74,7 +78,7 @@ def dao_update_organization(organization_id, **kwargs):
num_updated = db.session.execute(stmt).rowcount
if isinstance(domains, list):
stmt = delete(Domain).filter_by(organization_id=organization_id)
stmt = delete(Domain).where(Domain.organization_id == organization_id)
db.session.execute(stmt)
db.session.bulk_save_objects(
[
@@ -108,7 +112,7 @@ def _update_organization_services(organization, attribute, only_where_none=True)
@autocommit
@version_class(Service)
def dao_add_service_to_organization(service, organization_id):
stmt = select(Organization).filter_by(id=organization_id)
stmt = select(Organization).where(Organization.id == organization_id)
organization = db.session.execute(stmt).scalars().one()
service.organization_id = organization_id
@@ -121,7 +125,7 @@ def dao_get_users_for_organization(organization_id):
return (
db.session.query(User)
.join(User.organizations)
.filter(Organization.id == organization_id, User.state == "active")
.where(Organization.id == organization_id, User.state == "active")
.order_by(User.created_at)
.all()
)
@@ -130,7 +134,7 @@ 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)
stmt = select(User).filter_by(id=user_id)
stmt = select(User).where(User.id == user_id)
user = db.session.execute(stmt).scalars().one()
user.organizations.append(organization)
db.session.add(organization)

46
app/dao/permissions_dao.py
View File

@@ -1,7 +1,9 @@
from sqlalchemy import delete, select
from app import db
from app.dao import DAOClass
from app.enums import PermissionType
from app.models import Permission
from app.models import Permission, Service
class PermissionDAO(DAOClass):
@@ -14,22 +16,29 @@ class PermissionDAO(DAOClass):
self.create_instance(permission, _commit=False)
def remove_user_service_permissions(self, user, service):
query = self.Meta.model.query.filter_by(user=user, service=service)
query.delete()
db.session.execute(
delete(self.Meta.model).where(
self.Meta.model.user == user, self.Meta.model.service == service
)
)
db.session.commit()
def remove_user_service_permissions_for_all_services(self, user):
query = self.Meta.model.query.filter_by(user=user)
query.delete()
db.session.execute(delete(self.Meta.model).where(self.Meta.model.user == user))
db.session.commit()
def set_user_service_permission(
self, user, service, permissions, _commit=False, replace=False
):
try:
if replace:
query = self.Meta.model.query.filter(
self.Meta.model.user == user, self.Meta.model.service == service
db.session.execute(
delete(self.Meta.model).where(
self.Meta.model.user == user, self.Meta.model.service == service
)
)
query.delete()
db.session.commit()
for p in permissions:
p.user = user
p.service = service
@@ -44,17 +53,26 @@ class PermissionDAO(DAOClass):
def get_permissions_by_user_id(self, user_id):
return (
self.Meta.model.query.filter_by(user_id=user_id)
.join(Permission.service)
.filter_by(active=True)
db.session.execute(
select(Permission)
.join(Service)
.where(Permission.user_id == user_id)
.where(Service.active.is_(True))
)
.scalars()
.all()
)
def get_permissions_by_user_id_and_service_id(self, user_id, service_id):
return (
self.Meta.model.query.filter_by(user_id=user_id)
.join(Permission.service)
.filter_by(active=True, id=service_id)
db.session.execute(
select(Permission)
.join(Service)
.where(Permission.user_id == user_id)
.where(Service.active.is_(True))
.where(Service.id == service_id)
)
.scalars()
.all()
)

8
app/dao/provider_details_dao.py
View File

@@ -102,14 +102,14 @@ def dao_get_provider_stats():
current_datetime = utc_now()
first_day_of_the_month = current_datetime.date().replace(day=1)
subquery = (
substmt = (
db.session.query(
FactBilling.provider,
func.sum(FactBilling.billable_units * FactBilling.rate_multiplier).label(
"current_month_billable_sms"
),
)
.filter(
.where(
FactBilling.notification_type == NotificationType.SMS,
FactBilling.local_date >= first_day_of_the_month,
)
@@ -127,11 +127,11 @@ def dao_get_provider_stats():
ProviderDetails.updated_at,
ProviderDetails.supports_international,
User.name.label("created_by_name"),
func.coalesce(subquery.c.current_month_billable_sms, 0).label(
func.coalesce(substmt.c.current_month_billable_sms, 0).label(
"current_month_billable_sms"
),
)
.outerjoin(subquery, ProviderDetails.identifier == subquery.c.provider)
.outerjoin(substmt, ProviderDetails.identifier == substmt.c.provider)
.outerjoin(User, ProviderDetails.created_by_id == User.id)
.order_by(
ProviderDetails.notification_type,

43
app/dao/service_callback_api_dao.py
View File

@@ -1,3 +1,5 @@
from sqlalchemy import select
from app import create_uuid, db
from app.dao.dao_utils import autocommit, version_class
from app.enums import CallbackType
@@ -29,23 +31,42 @@ def reset_service_callback_api(
def get_service_callback_api(service_callback_api_id, service_id):
return ServiceCallbackApi.query.filter_by(
id=service_callback_api_id, service_id=service_id
).first()
return (
db.session.execute(
select(ServiceCallbackApi).where(
ServiceCallbackApi.id == service_callback_api_id,
ServiceCallbackApi.service_id == service_id,
)
)
.scalars()
.first()
)
def get_service_delivery_status_callback_api_for_service(service_id):
return ServiceCallbackApi.query.filter_by(
service_id=service_id,
callback_type=CallbackType.DELIVERY_STATUS,
).first()
return (
db.session.execute(
select(ServiceCallbackApi).where(
ServiceCallbackApi.service_id == service_id,
ServiceCallbackApi.callback_type == CallbackType.DELIVERY_STATUS,
)
)
.scalars()
.first()
)
def get_service_complaint_callback_api_for_service(service_id):
return ServiceCallbackApi.query.filter_by(
service_id=service_id,
callback_type=CallbackType.COMPLAINT,
).first()
return (
db.session.execute(
select(ServiceCallbackApi).where(
ServiceCallbackApi.service_id == service_id,
ServiceCallbackApi.callback_type == CallbackType.COMPLAINT,
)
)
.scalars()
.first()
)
@autocommit

21
app/dao/service_email_reply_to_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
@@ -10,7 +10,7 @@ from app.models import ServiceEmailReplyTo
def dao_get_reply_to_by_service_id(service_id):
reply_to = (
db.session.query(ServiceEmailReplyTo)
.filter(
.where(
ServiceEmailReplyTo.service_id == service_id,
ServiceEmailReplyTo.archived == False, # noqa
)
@@ -25,7 +25,7 @@ def dao_get_reply_to_by_service_id(service_id):
def dao_get_reply_to_by_id(service_id, reply_to_id):
reply_to = (
db.session.query(ServiceEmailReplyTo)
.filter(
.where(
ServiceEmailReplyTo.service_id == service_id,
ServiceEmailReplyTo.id == reply_to_id,
ServiceEmailReplyTo.archived == False, # noqa
@@ -62,7 +62,7 @@ def update_reply_to_email_address(service_id, reply_to_id, email_address, is_def
"You must have at least one reply to email address as the default.", 400
)
reply_to_update = ServiceEmailReplyTo.query.get(reply_to_id)
reply_to_update = db.session.get(ServiceEmailReplyTo, reply_to_id)
reply_to_update.email_address = email_address
reply_to_update.is_default = is_default
db.session.add(reply_to_update)
@@ -71,9 +71,16 @@ def update_reply_to_email_address(service_id, reply_to_id, email_address, is_def
@autocommit
def archive_reply_to_email_address(service_id, reply_to_id):
reply_to_archive = ServiceEmailReplyTo.query.filter_by(
id=reply_to_id, service_id=service_id
).one()
reply_to_archive = (
db.session.execute(
select(ServiceEmailReplyTo).where(
ServiceEmailReplyTo.id == reply_to_id,
ServiceEmailReplyTo.service_id == service_id,
)
)
.scalars()
.one()
)
if reply_to_archive.is_default:
raise ArchiveValidationError(

23
app/dao/service_inbound_api_dao.py
View File

@@ -1,3 +1,5 @@
from sqlalchemy import select
from app import create_uuid, db
from app.dao.dao_utils import autocommit, version_class
from app.models import ServiceInboundApi
@@ -28,13 +30,26 @@ def reset_service_inbound_api(
def get_service_inbound_api(service_inbound_api_id, service_id):
return ServiceInboundApi.query.filter_by(
id=service_inbound_api_id, service_id=service_id
).first()
return (
db.session.execute(
select(ServiceInboundApi).where(
ServiceInboundApi.id == service_inbound_api_id,
ServiceInboundApi.service_id == service_id,
)
)
.scalars()
.first()
)
def get_service_inbound_api_for_service(service_id):
return ServiceInboundApi.query.filter_by(service_id=service_id).first()
return (
db.session.execute(
select(ServiceInboundApi).where(ServiceInboundApi.service_id == service_id)
)
.scalars()
.first()
)
@autocommit

2
app/dao/service_permissions_dao.py
View File

@@ -7,7 +7,7 @@ from app.models import ServicePermission
def dao_fetch_service_permissions(service_id):
stmt = select(ServicePermission).filter(ServicePermission.service_id == service_id)
stmt = select(ServicePermission).where(ServicePermission.service_id == service_id)
return db.session.execute(stmt).scalars().all()

26
app/dao/service_sms_sender_dao.py
View File

@@ -17,8 +17,10 @@ def insert_service_sms_sender(service, sms_sender):
def dao_get_service_sms_senders_by_id(service_id, service_sms_sender_id):
stmt = select(ServiceSmsSender).filter_by(
id=service_sms_sender_id, service_id=service_id, archived=False
stmt = select(ServiceSmsSender).where(
ServiceSmsSender.id == service_sms_sender_id,
ServiceSmsSender.service_id == service_id,
ServiceSmsSender.archived == False, # noqa
)
return db.session.execute(stmt).scalars().one()
@@ -27,7 +29,10 @@ def dao_get_sms_senders_by_service_id(service_id):
stmt = (
select(ServiceSmsSender)
.filter_by(service_id=service_id, archived=False)
.where(
ServiceSmsSender.service_id == service_id,
ServiceSmsSender.archived == False, # noqa
)
.order_by(desc(ServiceSmsSender.is_default))
)
return db.session.execute(stmt).scalars().all()
@@ -65,7 +70,7 @@ def dao_update_service_sms_sender(
if old_default.id == service_sms_sender_id:
raise Exception("You must have at least one SMS sender as the default")
sms_sender_to_update = ServiceSmsSender.query.get(service_sms_sender_id)
sms_sender_to_update = db.session.get(ServiceSmsSender, service_sms_sender_id)
sms_sender_to_update.is_default = is_default
if not sms_sender_to_update.inbound_number_id and sms_sender:
sms_sender_to_update.sms_sender = sms_sender
@@ -85,9 +90,16 @@ def update_existing_sms_sender_with_inbound_number(
@autocommit
def archive_sms_sender(service_id, sms_sender_id):
sms_sender_to_archive = ServiceSmsSender.query.filter_by(
id=sms_sender_id, service_id=service_id
).one()
sms_sender_to_archive = (
db.session.execute(
select(ServiceSmsSender).where(
ServiceSmsSender.id == sms_sender_id,
ServiceSmsSender.service_id == service_id,
)
)
.scalars()
.one()
)
if sms_sender_to_archive.inbound_number_id:
raise ArchiveValidationError("You cannot delete an inbound number")

12
app/dao/service_user_dao.py
View File

@@ -6,7 +6,9 @@ from app.models import ServiceUser, User
def dao_get_service_user(user_id, service_id):
stmt = select(ServiceUser).filter_by(user_id=user_id, service_id=service_id)
stmt = select(ServiceUser).where(
ServiceUser.user_id == user_id, ServiceUser.service_id == service_id
)
return db.session.execute(stmt).scalars().one_or_none()
@@ -15,13 +17,17 @@ def dao_get_active_service_users(service_id):
stmt = (
select(ServiceUser)
.join(User, User.id == ServiceUser.user_id)
.filter(User.state == "active", ServiceUser.service_id == service_id)
.where(User.state == "active", ServiceUser.service_id == service_id)
)
return db.session.execute(stmt).scalars().all()
def dao_get_service_users_by_user_id(user_id):
return ServiceUser.query.filter_by(user_id=user_id).all()
return (
db.session.execute(select(ServiceUser).where(ServiceUser.user_id == user_id))
.scalars()
.all()
)
@autocommit

112
app/dao/services_dao.py
View File

@@ -96,7 +96,7 @@ def dao_fetch_live_services_data():
this_year_ft_billing = (
select(FactBilling)
.filter(
.where(
FactBilling.local_date >= year_start_date,
FactBilling.local_date <= year_end_date,
)
@@ -145,7 +145,7 @@ def dao_fetch_live_services_data():
this_year_ft_billing, Service.id == this_year_ft_billing.c.service_id
)
.outerjoin(User, Service.go_live_user_id == User.id)
.filter(
.where(
Service.count_as_live.is_(True),
Service.active.is_(True),
Service.restricted.is_(False),
@@ -216,10 +216,12 @@ def dao_fetch_service_by_inbound_number(number):
def dao_fetch_service_by_id_with_api_keys(service_id, only_active=False):
stmt = (
select(Service).filter_by(id=service_id).options(joinedload(Service.api_keys))
select(Service)
.where(Service.id == service_id)
.options(joinedload(Service.api_keys))
)
if only_active:
stmt = stmt.filter(Service.active)
stmt = stmt.where(Service.active)
return db.session.execute(stmt).scalars().unique().one()
@@ -227,12 +229,12 @@ def dao_fetch_all_services_by_user(user_id, only_active=False):
stmt = (
select(Service)
.filter(Service.users.any(id=user_id))
.where(Service.users.any(id=user_id))
.order_by(asc(Service.created_at))
.options(joinedload(Service.users))
)
if only_active:
stmt = stmt.filter(Service.active)
stmt = stmt.where(Service.active)
return db.session.execute(stmt).scalars().unique().all()
@@ -240,7 +242,7 @@ def dao_fetch_all_services_created_by_user(user_id):
stmt = (
select(Service)
.filter_by(created_by_id=user_id)
.where(Service.created_by_id == user_id)
.order_by(asc(Service.created_at))
)
@@ -260,7 +262,7 @@ def dao_archive_service(service_id):
joinedload(Service.templates).subqueryload(Template.template_redacted),
joinedload(Service.api_keys),
)
.filter(Service.id == service_id)
.where(Service.id == service_id)
)
service = db.session.execute(stmt).scalars().unique().one()
@@ -281,7 +283,7 @@ def dao_fetch_service_by_id_and_user(service_id, user_id):
stmt = (
select(Service)
.filter(Service.users.any(id=user_id), Service.id == service_id)
.where(Service.users.any(id=user_id), Service.id == service_id)
.options(joinedload(Service.users))
)
result = db.session.execute(stmt).scalar_one()
@@ -392,27 +394,39 @@ def delete_service_and_all_associated_db_objects(service):
db.session.execute(stmt)
db.session.commit()
subq = select(Template.id).filter_by(service=service).subquery()
subq = select(Template.id).where(Template.service == service).subquery()
stmt = delete(TemplateRedacted).filter(TemplateRedacted.template_id.in_(subq))
stmt = delete(TemplateRedacted).where(TemplateRedacted.template_id.in_(subq))
_delete_commit(stmt)
_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))
_delete_commit(delete(ServiceSmsSender).where(ServiceSmsSender.service == service))
_delete_commit(
delete(ServiceEmailReplyTo).where(ServiceEmailReplyTo.service == service)
)
_delete_commit(delete(InvitedUser).where(InvitedUser.service == service))
_delete_commit(delete(Permission).where(Permission.service == service))
_delete_commit(
delete(NotificationHistory).where(NotificationHistory.service == service)
)
_delete_commit(delete(Notification).where(Notification.service == service))
_delete_commit(delete(Job).where(Job.service == service))
_delete_commit(delete(Template).where(Template.service == service))
_delete_commit(
delete(TemplateHistory).where(TemplateHistory.service_id == service.id)
)
_delete_commit(
delete(ServicePermission).where(ServicePermission.service_id == service.id)
)
_delete_commit(delete(ApiKey).where(ApiKey.service == service))
_delete_commit(
delete(ApiKey.get_history_model()).where(
ApiKey.get_history_model().service_id == service.id
)
)
_delete_commit(delete(AnnualBilling).where(AnnualBilling.service_id == service.id))
stmt = (
select(VerifyCode).join(User).filter(User.id.in_([x.id for x in service.users]))
select(VerifyCode).join(User).where(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))
@@ -421,7 +435,7 @@ def delete_service_and_all_associated_db_objects(service):
for user in users:
user.organizations = []
service.users.remove(user)
_delete_commit(delete(Service.get_history_model()).filter_by(id=service.id))
_delete_commit(delete(Service.get_history_model()).where(Service.id == service.id))
db.session.delete(service)
db.session.commit()
for user in users:
@@ -438,7 +452,7 @@ def dao_fetch_todays_stats_for_service(service_id):
Notification.status,
func.count(Notification.id).label("count"),
)
.filter(
.where(
Notification.service_id == service_id,
Notification.key_type != KeyType.TEST,
Notification.created_at >= start_date,
@@ -578,14 +592,14 @@ def dao_fetch_todays_stats_for_all_services(
start_date = get_midnight_in_utc(today)
end_date = get_midnight_in_utc(today + timedelta(days=1))
subquery = (
substmt = (
select(
Notification.notification_type,
Notification.status,
Notification.service_id,
func.count(Notification.id).label("count"),
)
.filter(
.where(
Notification.created_at >= start_date, Notification.created_at < end_date
)
.group_by(
@@ -594,9 +608,9 @@ def dao_fetch_todays_stats_for_all_services(
)
if not include_from_test_key:
subquery = subquery.filter(Notification.key_type != KeyType.TEST)
substmt = substmt.where(Notification.key_type != KeyType.TEST)
subquery = subquery.subquery()
substmt = substmt.subquery()
stmt = (
select(
@@ -605,16 +619,16 @@ def dao_fetch_todays_stats_for_all_services(
Service.restricted,
Service.active,
Service.created_at,
subquery.c.notification_type,
subquery.c.status,
subquery.c.count,
substmt.c.notification_type,
substmt.c.status,
substmt.c.count,
)
.outerjoin(subquery, subquery.c.service_id == Service.id)
.outerjoin(substmt, substmt.c.service_id == Service.id)
.order_by(Service.id)
)
if only_active:
stmt = stmt.filter(Service.active)
stmt = stmt.where(Service.active)
return db.session.execute(stmt).all()
@@ -629,7 +643,7 @@ def dao_suspend_service(service_id):
stmt = (
select(Service)
.options(joinedload(Service.api_keys))
.filter(Service.id == service_id)
.where(Service.id == service_id)
)
service = db.session.execute(stmt).scalars().unique().one()
@@ -662,7 +676,7 @@ def dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=500)
Notification.service_id.label("service_id"),
func.count(Notification.id).label("notification_count"),
)
.filter(
.where(
Notification.service_id == Service.id,
Notification.created_at >= start_date,
Notification.created_at <= end_date,
@@ -681,12 +695,12 @@ def dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=500)
def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10000):
subquery = (
substmt = (
select(
func.count(Notification.id).label("total_count"),
Notification.service_id.label("service_id"),
)
.filter(
.where(
Notification.service_id == Service.id,
Notification.created_at >= start_date,
Notification.created_at <= end_date,
@@ -701,20 +715,20 @@ def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10
.having(func.count(Notification.id) >= threshold)
)
subquery = subquery.subquery()
substmt = substmt.subquery()
stmt = (
select(
Notification.service_id.label("service_id"),
func.count(Notification.id).label("permanent_failure_count"),
subquery.c.total_count.label("total_count"),
substmt.c.total_count.label("total_count"),
(
cast(func.count(Notification.id), Float)
/ cast(subquery.c.total_count, Float)
/ cast(substmt.c.total_count, Float)
).label("permanent_failure_rate"),
)
.join(subquery, subquery.c.service_id == Notification.service_id)
.filter(
.join(substmt, substmt.c.service_id == Notification.service_id)
.where(
Notification.service_id == Service.id,
Notification.created_at >= start_date,
Notification.created_at <= end_date,
@@ -724,10 +738,10 @@ def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10
Service.restricted == False, # noqa
Service.active == True, # noqa
)
.group_by(Notification.service_id, subquery.c.total_count)
.group_by(Notification.service_id, substmt.c.total_count)
.having(
cast(func.count(Notification.id), Float)
/ cast(subquery.c.total_count, Float)
/ cast(substmt.c.total_count, Float)
>= 0.25
)
)
@@ -746,7 +760,7 @@ def get_live_services_with_organization():
)
.select_from(Service)
.outerjoin(Service.organization)
.filter(
.where(
Service.count_as_live.is_(True),
Service.active.is_(True),
Service.restricted.is_(False),
@@ -768,7 +782,7 @@ def fetch_notification_stats_for_service_by_month_by_user(
(NotificationAllTimeView.status).label("notification_status"),
func.count(NotificationAllTimeView.id).label("count"),
)
.filter(
.where(
NotificationAllTimeView.service_id == service_id,
NotificationAllTimeView.created_at >= start_date,
NotificationAllTimeView.created_at < end_date,

4
app/dao/template_folder_dao.py
View File

@@ -6,14 +6,14 @@ from app.models import TemplateFolder
def dao_get_template_folder_by_id_and_service_id(template_folder_id, service_id):
stmt = select(TemplateFolder).filter(
stmt = select(TemplateFolder).where(
TemplateFolder.id == template_folder_id, TemplateFolder.service_id == service_id
)
return db.session.execute(stmt).scalars().one()
def dao_get_valid_template_folders_by_id(folder_ids):
stmt = select(TemplateFolder).filter(TemplateFolder.id.in_(folder_ids))
stmt = select(TemplateFolder).where(TemplateFolder.id.in_(folder_ids))
return db.session.execute(stmt).scalars().all()

43
app/dao/templates_dao.py
View File

@@ -46,21 +46,28 @@ 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:
stmt = select(TemplateHistory).filter_by(
id=template_id, hidden=False, service_id=service_id, version=version
stmt = select(TemplateHistory).where(
TemplateHistory.id == template_id,
TemplateHistory.hidden == False, # noqa
TemplateHistory.service_id == service_id,
TemplateHistory.version == version,
)
return db.session.execute(stmt).scalars().one()
stmt = select(Template).filter_by(
id=template_id, hidden=False, service_id=service_id
stmt = select(Template).where(
Template.id == template_id,
Template.hidden == False, # noqa
Template.service_id == service_id,
)
return db.session.execute(stmt).scalars().one()
def dao_get_template_by_id(template_id, version=None):
if version is not None:
stmt = select(TemplateHistory).filter_by(id=template_id, version=version)
stmt = select(TemplateHistory).where(
TemplateHistory.id == template_id, TemplateHistory.version == version
)
return db.session.execute(stmt).scalars().one()
stmt = select(Template).filter_by(id=template_id)
stmt = select(Template).where(Template.id == template_id)
return db.session.execute(stmt).scalars().one()
@@ -68,11 +75,11 @@ def dao_get_all_templates_for_service(service_id, template_type=None):
if template_type is not None:
stmt = (
select(Template)
.filter_by(
service_id=service_id,
template_type=template_type,
hidden=False,
archived=False,
.where(
Template.service_id == service_id,
Template.template_type == template_type,
Template.hidden == False, # noqa
Template.archived == False, # noqa
)
.order_by(
asc(Template.name),
@@ -82,7 +89,11 @@ def dao_get_all_templates_for_service(service_id, template_type=None):
return db.session.execute(stmt).scalars().all()
stmt = (
select(Template)
.filter_by(service_id=service_id, hidden=False, archived=False)
.where(
Template.service_id == service_id,
Template.hidden == False, # noqa
Template.archived == False, # noqa
)
.order_by(
asc(Template.name),
asc(Template.template_type),
@@ -94,10 +105,10 @@ def dao_get_all_templates_for_service(service_id, template_type=None):
def dao_get_template_versions(service_id, template_id):
stmt = (
select(TemplateHistory)
.filter_by(
service_id=service_id,
id=template_id,
hidden=False,
.where(
TemplateHistory.service_id == service_id,
TemplateHistory.id == template_id,
TemplateHistory.hidden == False, # noqa
)
.order_by(desc(TemplateHistory.version))
)

64
app/dao/uploads_dao.py
View File

@@ -1,9 +1,10 @@
from os import getenv
from flask import current_app
from sqlalchemy import String, and_, desc, func, literal, text
from sqlalchemy import String, and_, desc, func, literal, select, text, union
from app import db
from app.dao.inbound_sms_dao import Pagination
from app.enums import JobStatus, NotificationStatus, NotificationType
from app.models import Job, Notification, ServiceDataRetention, Template
from app.utils import midnight_n_days_ago, utc_now
@@ -51,8 +52,8 @@ def dao_get_uploads_by_service_id(service_id, limit_days=None, page=1, page_size
if limit_days is not None:
jobs_query_filter.append(Job.created_at >= midnight_n_days_ago(limit_days))
jobs_query = (
db.session.query(
jobs_stmt = (
select(
Job.id,
Job.original_file_name,
Job.notification_count,
@@ -67,6 +68,7 @@ def dao_get_uploads_by_service_id(service_id, limit_days=None, page=1, page_size
literal("job").label("upload_type"),
literal(None).label("recipient"),
)
.select_from(Job)
.join(Template, Job.template_id == Template.id)
.outerjoin(
ServiceDataRetention,
@@ -76,7 +78,7 @@ def dao_get_uploads_by_service_id(service_id, limit_days=None, page=1, page_size
== func.cast(ServiceDataRetention.notification_type, String),
),
)
.filter(*jobs_query_filter)
.where(*jobs_query_filter)
)
letters_query_filter = [
@@ -93,13 +95,14 @@ def dao_get_uploads_by_service_id(service_id, limit_days=None, page=1, page_size
Notification.created_at >= midnight_n_days_ago(limit_days)
)
letters_subquery = (
db.session.query(
letters_substmt = (
select(
func.count().label("notification_count"),
_naive_gmt_to_utc(_get_printing_datetime(Notification.created_at)).label(
"printing_at"
),
)
.select_from(Notification)
.join(Template, Notification.template_id == Template.id)
.outerjoin(
ServiceDataRetention,
@@ -109,30 +112,39 @@ def dao_get_uploads_by_service_id(service_id, limit_days=None, page=1, page_size
== func.cast(ServiceDataRetention.notification_type, String),
),
)
.filter(*letters_query_filter)
.where(*letters_query_filter)
.group_by("printing_at")
.subquery()
)
letters_query = db.session.query(
literal(None).label("id"),
literal("Uploaded letters").label("original_file_name"),
letters_subquery.c.notification_count.label("notification_count"),
literal("letter").label("template_type"),
literal(None).label("days_of_retention"),
letters_subquery.c.printing_at.label("created_at"),
literal(None).label("scheduled_for"),
letters_subquery.c.printing_at.label("processing_started"),
literal(None).label("status"),
literal("letter_day").label("upload_type"),
literal(None).label("recipient"),
).group_by(
letters_subquery.c.notification_count,
letters_subquery.c.printing_at,
letters_stmt = (
select(
literal(None).label("id"),
literal("Uploaded letters").label("original_file_name"),
letters_substmt.c.notification_count.label("notification_count"),
literal("letter").label("template_type"),
literal(None).label("days_of_retention"),
letters_substmt.c.printing_at.label("created_at"),
literal(None).label("scheduled_for"),
letters_substmt.c.printing_at.label("processing_started"),
literal(None).label("status"),
literal("letter_day").label("upload_type"),
literal(None).label("recipient"),
)
.select_from(Notification)
.group_by(
letters_substmt.c.notification_count,
letters_substmt.c.printing_at,
)
)
return (
jobs_query.union_all(letters_query)
.order_by(desc("processing_started"), desc("created_at"))
.paginate(page=page, per_page=page_size)
stmt = union(jobs_stmt, letters_stmt).order_by(
desc("processing_started"), desc("created_at")
)
results = db.session.execute(stmt).all()
page_size = current_app.config["PAGE_SIZE"]
offset = (page - 1) * page_size
paginated_results = results[offset : offset + page_size]
pagination = Pagination(paginated_results, page, page_size, len(results))
return pagination

24
app/dao/users_dao.py
View File

@@ -37,7 +37,7 @@ def get_login_gov_user(login_uuid, email_address):
login.gov uuids are. Eventually the code that checks by email address
should be removed.
"""
stmt = select(User).filter_by(login_uuid=login_uuid)
stmt = select(User).where(User.login_uuid == login_uuid)
user = db.session.execute(stmt).scalars().first()
if user:
if user.email_address != email_address:
@@ -54,7 +54,7 @@ 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
stmt = select(User).filter(User.email_address.ilike(email_address))
stmt = select(User).where(User.email_address.ilike(email_address))
user = db.session.execute(stmt).scalars().first()
if user:
@@ -65,7 +65,7 @@ def get_login_gov_user(login_uuid, email_address):
def save_user_attribute(usr, update_dict=None):
db.session.query(User).filter_by(id=usr.id).update(update_dict or {})
db.session.query(User).where(User.id == usr.id).update(update_dict or {})
db.session.commit()
@@ -82,7 +82,7 @@ def save_model_user(
user.email_access_validated_at = utc_now()
if update_dict:
_remove_values_for_keys_if_present(update_dict, ["id", "password_changed_at"])
db.session.query(User).filter_by(id=user.id).update(update_dict or {})
db.session.query(User).where(User.id == user.id).update(update_dict or {})
else:
db.session.add(user)
db.session.commit()
@@ -105,7 +105,7 @@ def get_user_code(user, code, code_type):
# time searching for the correct code.
stmt = (
select(VerifyCode)
.filter_by(user=user, code_type=code_type)
.where(VerifyCode.user == user, VerifyCode.code_type == code_type)
.order_by(VerifyCode.created_at.desc())
)
codes = db.session.execute(stmt).scalars().all()
@@ -113,7 +113,7 @@ def get_user_code(user, code, code_type):
def delete_codes_older_created_more_than_a_day_ago():
stmt = delete(VerifyCode).filter(
stmt = delete(VerifyCode).where(
VerifyCode.created_at < utc_now() - timedelta(hours=24)
)
@@ -135,13 +135,13 @@ def delete_model_user(user):
def delete_user_verify_codes(user):
stmt = delete(VerifyCode).filter_by(user=user)
stmt = delete(VerifyCode).where(VerifyCode.user == user)
db.session.execute(stmt)
db.session.commit()
def count_user_verify_codes(user):
stmt = select(func.count(VerifyCode.id)).filter(
stmt = select(func.count(VerifyCode.id)).where(
VerifyCode.user == user,
VerifyCode.expiry_datetime > utc_now(),
VerifyCode.code_used.is_(False),
@@ -152,7 +152,7 @@ def count_user_verify_codes(user):
def get_user_by_id(user_id=None):
if user_id:
stmt = select(User).filter_by(id=user_id)
stmt = select(User).where(User.id == user_id)
return db.session.execute(stmt).scalars().one()
return get_users()
@@ -163,13 +163,13 @@ def get_users():
def get_user_by_email(email):
stmt = select(User).filter(func.lower(User.email_address) == func.lower(email))
stmt = select(User).where(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)
stmt = select(User).filter(User.email_address.ilike("%{}%".format(email)))
stmt = select(User).where(User.email_address.ilike("%{}%".format(email)))
return db.session.execute(stmt).scalars().all()
@@ -200,7 +200,7 @@ def get_user_and_accounts(user_id):
# that we have put is functionally doing the same thing as before
stmt = (
select(User)
.filter(User.id == user_id)
.where(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)