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

I think it is working now.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-08-28 15:05:42 -04:00
parent fb743cf28b
commit 4084a18327
2 changed files with 207 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
app/dao/fact_notification_status_dao.py
View File

@@ -1,6 +1,6 @@
from datetime import date, timedelta from datetime import timedelta
from sqlalchemy import Date, case, func, select, union_all, cast from sqlalchemy import Date, case, cast, func, select, union_all
from sqlalchemy.dialects.postgresql import insert from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import aliased from sqlalchemy.orm import aliased
from sqlalchemy.sql.expression import extract, literal from sqlalchemy.sql.expression import extract, literal
@@ -130,12 +130,10 @@ def fetch_notification_status_for_service_for_day(fetch_day, service_id):
def fetch_notification_status_for_service_for_today_and_7_previous_days( def fetch_notification_status_for_service_for_today_and_7_previous_days(
service_id: str, service_id: str, by_template: bool = False, limit_days: int = 7
by_template: bool = False,
limit_days: int = 7
) -> list[dict | None]: ) -> list[dict | None]:
start_date = midnight_n_days_ago(limit_days) start_date = midnight_n_days_ago(limit_days)
now = utc_now() now = get_midnight_in_utc(utc_now())
# Query for the last 7 days # Query for the last 7 days
stats_for_7_days = select( stats_for_7_days = select(
@@ -145,7 +143,9 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(
[ [
FactNotificationStatus.template_id.label("template_id"), FactNotificationStatus.template_id.label("template_id"),
FactNotificationStatus.local_date.label("date_used"), FactNotificationStatus.local_date.label("date_used"),
] if by_template else [] ]
if by_template
else []
), ),
FactNotificationStatus.notification_count.label("count"), FactNotificationStatus.notification_count.label("count"),
).where( ).where(
@@ -155,41 +155,53 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(
) )
# Query for today's stats # Query for today's stats
stats_for_today = select( stats_for_today = (
cast(Notification.notification_type, Text), select(
cast(Notification.status, Text), cast(Notification.notification_type, Text),
*( cast(Notification.status, Text),
[ *(
Notification.template_id, [
literal(date.today()).label("date_used"), Notification.template_id,
] if by_template else [] literal(now).label("date_used"),
), ]
func.count().label("count"), if by_template
).where( else []
Notification.created_at >= get_midnight_in_utc(now), ),
Notification.service_id == service_id, func.count().label("count"),
Notification.key_type != KeyType.TEST, )
).group_by( .where(
Notification.notification_type, Notification.created_at >= now,
*([Notification.template_id] if by_template else []), Notification.service_id == service_id,
Notification.status, Notification.key_type != KeyType.TEST,
)
.group_by(
Notification.notification_type,
*([Notification.template_id] if by_template else []),
Notification.status,
)
) )
# Combine the queries using union_all # Combine the queries using union_all
all_stats_union = union_all(stats_for_7_days, stats_for_today).subquery() all_stats_union = union_all(stats_for_7_days, stats_for_today).subquery()
all_stats_alias = aliased(all_stats_union) all_stats_alias = aliased(all_stats_union, name="all_stats")
# Final query with optional template joins # Final query with optional template joins
query = select( query = select(
*( *(
[ [
TemplateFolder.name.label("folder"),
Template.name.label("template_name"), Template.name.label("template_name"),
False, # TODO: Handle `is_precompiled_letter` False, # TODO: Handle `is_precompiled_letter`
template_folder_map.c.template_folder_id,
all_stats_alias.c.template_id, all_stats_alias.c.template_id,
TemplateFolder.name.label("folder"),
User.name.label("created_by"), User.name.label("created_by"),
func.max(all_stats_alias.c.date_used).label("last_used"), # Get the most recent date Template.created_by_id,
] if by_template else [] func.max(all_stats_alias.c.date_used).label(
"last_used"
), # Get the most recent date
]
if by_template
else []
), ),
all_stats_alias.c.notification_type, all_stats_alias.c.notification_type,
all_stats_alias.c.status, all_stats_alias.c.status,
@@ -197,38 +209,36 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(
) )
if by_template: if by_template:
query = query.join( query = (
Template, query.join(Template, all_stats_alias.c.template_id == Template.id)
all_stats_alias.c.template_id == Template.id .join(User, Template.created_by_id == User.id)
).join( .outerjoin(
User, template_folder_map, Template.id == template_folder_map.c.template_id
Template.created_by_id == User.id )
).join( .outerjoin(
template_folder_map, TemplateFolder,
Template.id == template_folder_map.c.template_id TemplateFolder.id == template_folder_map.c.template_folder_id,
).join( )
TemplateFolder,
TemplateFolder.id == template_folder_map.c.template_id
) )
# Group by all necessary fields except date_used # Group by all necessary fields except date_used
query = query.group_by( query = query.group_by(
*( *(
[ [
TemplateFolder.name,
Template.name, Template.name,
all_stats_alias.c.template_id, all_stats_alias.c.template_id,
TemplateFolder.name,
User.name, User.name,
] if by_template else [] template_folder_map.c.template_folder_id,
Template.created_by_id,
]
if by_template
else []
), ),
all_stats_alias.c.notification_type, all_stats_alias.c.notification_type,
all_stats_alias.c.status, all_stats_alias.c.status,
) )
print("*"*79)
print(query)
print("*"*79)
# Execute the query using Flask-SQLAlchemy's session # Execute the query using Flask-SQLAlchemy's session
result = db.session.execute(query) result = db.session.execute(query)
return result.mappings().all() return result.mappings().all()

227
tests/app/dao/test_fact_notification_status_dao.py
View File

@@ -1,5 +1,4 @@
from datetime import date, datetime, timedelta from datetime import date, datetime, timedelta
from unittest import mock
from uuid import UUID from uuid import UUID
import pytest import pytest
@@ -26,6 +25,7 @@ from tests.app.db import (
create_notification, create_notification,
create_service, create_service,
create_template, create_template,
create_template_folder,
) )
@@ -252,15 +252,18 @@ def test_fetch_notification_status_by_template_for_service_for_today_and_7_previ
notify_db_session, notify_db_session,
): ):
service_1 = create_service(service_name="service_1") service_1 = create_service(service_name="service_1")
test_folder = create_template_folder(service=service_1, name="Test_Folder_For_This")
sms_template = create_template( sms_template = create_template(
template_name="sms Template 1", template_name="sms Template 1",
service=service_1, service=service_1,
template_type=TemplateType.SMS, template_type=TemplateType.SMS,
folder=test_folder,
) )
sms_template_2 = create_template( sms_template_2 = create_template(
template_name="sms Template 2", template_name="sms Template 2",
service=service_1, service=service_1,
template_type=TemplateType.SMS, template_type=TemplateType.SMS,
folder=test_folder,
) )
email_template = create_template( email_template = create_template(
service=service_1, template_type=TemplateType.EMAIL service=service_1, template_type=TemplateType.EMAIL
@@ -329,82 +332,152 @@ def test_fetch_notification_status_by_template_for_service_for_today_and_7_previ
by_template=True, by_template=True,
) )
assert [ expected = [
( {
"email Template Name", "folder": None,
False, "template_name": "email Template Name",
mock.ANY, "_no_label": False,
NotificationType.EMAIL, "created_by": "Test User",
NotificationStatus.DELIVERED, "last_used": datetime(2018, 10, 31, 0, 0),
1, "notification_type": NotificationType.EMAIL,
), "status": NotificationStatus.DELIVERED,
( "count": 1,
"email Template Name", },
False, {
mock.ANY, "folder": None,
NotificationType.EMAIL, "template_name": "email Template Name",
NotificationStatus.DELIVERED, "_no_label": False,
3, "created_by": "Test User",
), "last_used": datetime(2018, 10, 29, 0, 0),
( "notification_type": NotificationType.EMAIL,
"sms Template 1", "status": NotificationStatus.DELIVERED,
False, "count": 3,
mock.ANY, },
NotificationType.SMS, {
NotificationStatus.CREATED, "folder": None,
1, "template_name": "sms Template Name",
), "_no_label": False,
( "created_by": "Test User",
"sms Template Name", "last_used": datetime(2018, 10, 29, 0, 0),
False, "notification_type": NotificationType.SMS,
mock.ANY, "status": NotificationStatus.CREATED,
NotificationType.SMS, "count": 1,
NotificationStatus.CREATED, },
1, {
), "folder": "Test_Folder_For_This",
( "template_name": "sms Template 1",
"sms Template 1", "_no_label": False,
False, "created_by": "Test User",
mock.ANY, "last_used": datetime(2018, 10, 31, 0, 0),
NotificationType.SMS, "notification_type": NotificationType.SMS,
NotificationStatus.DELIVERED, "status": NotificationStatus.CREATED,
1, "count": 1,
), },
( {
"sms Template 2", "folder": None,
False, "template_name": "sms Template Name",
mock.ANY, "_no_label": False,
NotificationType.SMS, "created_by": "Test User",
NotificationStatus.DELIVERED, "last_used": datetime(2018, 10, 29, 0, 0),
1, "notification_type": NotificationType.SMS,
), "status": NotificationStatus.DELIVERED,
( "count": 10,
"sms Template Name", },
False, {
mock.ANY, "folder": "Test_Folder_For_This",
NotificationType.SMS, "template_name": "sms Template 2",
NotificationStatus.DELIVERED, "_no_label": False,
8, "created_by": "Test User",
), "last_used": datetime(2018, 10, 31, 0, 0),
( "notification_type": NotificationType.SMS,
"sms Template Name", "status": NotificationStatus.DELIVERED,
False, "count": 1,
mock.ANY, },
NotificationType.SMS, {
NotificationStatus.DELIVERED, "folder": None,
10, "template_name": "sms Template Name",
), "_no_label": False,
( "created_by": "Test User",
"sms Template Name", "last_used": datetime(2018, 10, 25, 0, 0),
False, "notification_type": NotificationType.SMS,
mock.ANY, "status": NotificationStatus.DELIVERED,
NotificationType.SMS, "count": 8,
NotificationStatus.DELIVERED, },
11, {
), "folder": "Test_Folder_For_This",
] == sorted( "template_name": "sms Template 1",
results, key=lambda x: (x.notification_type, x.status, x.template_name, x.count) "_no_label": False,
) "created_by": "Test User",
"last_used": datetime(2018, 10, 31, 0, 0),
"notification_type": NotificationType.SMS,
"status": NotificationStatus.DELIVERED,
"count": 1,
},
{
"folder": None,
"template_name": "sms Template Name",
"_no_label": False,
"created_by": "Test User",
"last_used": datetime(2018, 10, 29, 0, 0),
"notification_type": NotificationType.SMS,
"status": NotificationStatus.DELIVERED,
"count": 11,
},
]
expected = [
[
str(row[k]) if k != "last_used" else row[k].strftime("%Y-%m-%d")
for k in (
"folder",
"template_name",
"created_by",
"last_used",
"notification_type",
"status",
"count",
)
]
for row in sorted(
expected,
key=lambda x: (
str(x["notification_type"]),
str(x["status"]),
x["folder"] if x["folder"] is not None else "",
x["template_name"],
x["count"],
x["last_used"],
),
)
]
results = [
[
str(row[k]) if k != "last_used" else row[k].strftime("%Y-%m-%d")
for k in (
"folder",
"template_name",
"created_by",
"last_used",
"notification_type",
"status",
"count",
)
]
for row in sorted(
results,
key=lambda x: (
x.notification_type,
x.status,
x.folder if x.folder is not None else "",
x.template_name,
x.count,
x.last_used,
),
)
]
assert expected == results
@pytest.mark.parametrize( @pytest.mark.parametrize(