make create_nightly_notification_status_for_day take notification_type

the nightly task won't be affected, it'll just trigger three times more
sub-tasks.

this doesn't need to be a two-part deploy because we only trigger this
overnight, so as long as the deploy completes in daytime we don't need
to worry about celery task signatures
This commit is contained in:
Leo Hemsted
2019-12-05 12:11:30 +00:00
parent 30f361d318
commit 0448bca542
5 changed files with 82 additions and 51 deletions

View File

@@ -12,6 +12,11 @@ from app.dao.fact_billing_dao import (
update_fact_billing
)
from app.dao.fact_notification_status_dao import fetch_notification_status_for_day, update_fact_notification_status
from app.models import (
SMS_TYPE,
EMAIL_TYPE,
LETTER_TYPE,
)
@notify_celery.task(name="create-nightly-billing")
@@ -72,30 +77,31 @@ def create_nightly_notification_status(day_start=None):
day_start = datetime.strptime(day_start, "%Y-%m-%d").date()
for i in range(0, 4):
process_day = day_start - timedelta(days=i)
create_nightly_notification_status_for_day.apply_async(
kwargs={'process_day': process_day.isoformat()},
queue=QueueNames.REPORTING
)
for notification_type in [SMS_TYPE, EMAIL_TYPE, LETTER_TYPE]:
create_nightly_notification_status_for_day.apply_async(
kwargs={'process_day': process_day.isoformat(), 'notification_type': notification_type},
queue=QueueNames.REPORTING
)
@notify_celery.task(name="create-nightly-notification-status-for-day")
@statsd(namespace="tasks")
def create_nightly_notification_status_for_day(process_day):
def create_nightly_notification_status_for_day(process_day, notification_type):
process_day = datetime.strptime(process_day, "%Y-%m-%d").date()
start = datetime.utcnow()
transit_data = fetch_notification_status_for_day(process_day=process_day)
transit_data = fetch_notification_status_for_day(process_day=process_day, notification_type=notification_type)
end = datetime.utcnow()
current_app.logger.info('create-nightly-notification-status-for-day {} fetched in {} seconds'.format(
current_app.logger.info('create-nightly-notification-status-for-day {} type {} fetched in {} seconds'.format(
process_day,
notification_type,
(end - start).seconds)
)
update_fact_notification_status(transit_data, process_day)
update_fact_notification_status(transit_data, process_day, notification_type)
current_app.logger.info(
"create-nightly-notification-status-for-day task complete: {} rows updated for day: {}".format(
len(transit_data), process_day
"create-nightly-notification-status-for-day task complete: {} rows updated for type {} for day: {}".format(
len(transit_data), process_day, notification_type
)
)

View File

@@ -44,11 +44,19 @@ from app.dao.services_dao import (
from app.dao.templates_dao import dao_get_template_by_id
from app.dao.users_dao import delete_model_user, delete_user_verify_codes, get_user_by_email
from app.models import (
PROVIDERS, User, Notification, Organisation, Domain, Service, SMS_TYPE,
PROVIDERS,
NOTIFICATION_CREATED,
KEY_TYPE_TEST,
SMS_TYPE,
EMAIL_TYPE,
LETTER_TYPE,
User,
Notification,
Organisation,
Domain,
Service,
EmailBranding,
LetterBranding
LetterBranding,
)
from app.performance_platform.processing_time import send_processing_time_for_start_and_end
from app.utils import get_london_midnight_in_utc, get_midnight_for_day_before
@@ -490,16 +498,23 @@ def rebuild_ft_billing_for_day(service_id, day):
@notify_command(name='migrate-data-to-ft-notification-status')
@click.option('-s', '--start_date', required=True, help="start date inclusive", type=click_dt(format='%Y-%m-%d'))
@click.option('-e', '--end_date', required=True, help="end date inclusive", type=click_dt(format='%Y-%m-%d'))
@click.option('-t', '--notification-type', required=False, help="notification type (or leave blank for all types)")
@statsd(namespace="tasks")
def migrate_data_to_ft_notification_status(start_date, end_date):
def migrate_data_to_ft_notification_status(start_date, end_date, notification_type=None):
notification_types = [SMS_TYPE, LETTER_TYPE, EMAIL_TYPE] if notification_type is None else [notification_type]
start_date = start_date.date()
for day_diff in range((end_date - start_date).days):
process_day = start_date + timedelta(days=day_diff)
print('create_nightly_notification_status_for_day triggered for {}'.format(process_day))
create_nightly_notification_status_for_day.apply_async(
kwargs={'process_day': process_day.strftime('%Y-%m-%d')},
queue=QueueNames.REPORTING
)
for notification_type in notification_types:
print('create_nightly_notification_status_for_day triggered for {} and {}'.format(
process_day,
notification_type
))
create_nightly_notification_status_for_day.apply_async(
kwargs={'process_day': process_day.strftime('%Y-%m-%d'), 'notification_type': notification_type},
queue=QueueNames.REPORTING
)
@notify_command(name='bulk-invite-user-to-service')

View File

@@ -9,10 +9,8 @@ from sqlalchemy.types import DateTime, Integer
from app import db
from app.models import (
EMAIL_TYPE,
FactNotificationStatus,
KEY_TYPE_TEST,
LETTER_TYPE,
Notification,
NOTIFICATION_CANCELLED,
NOTIFICATION_CREATED,
@@ -24,7 +22,6 @@ from app.models import (
NOTIFICATION_TEMPORARY_FAILURE,
NOTIFICATION_PERMANENT_FAILURE,
Service,
SMS_TYPE,
Template,
)
from app.dao.dao_utils import transactional
@@ -36,7 +33,7 @@ from app.utils import (
)
def fetch_notification_status_for_day(process_day):
def fetch_notification_status_for_day(process_day, notification_type):
start_date = convert_bst_to_utc(datetime.combine(process_day, time.min))
end_date = convert_bst_to_utc(datetime.combine(process_day + timedelta(days=1), time.min))
@@ -44,23 +41,19 @@ def fetch_notification_status_for_day(process_day):
all_data_for_process_day = []
services = Service.query.all()
# for each service
# for each notification type
# query notifications for day
# if no rows try notificationHistory
# for each service query notifications or notification_history for the day, depending on their data retention
for service in services:
for notification_type in [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE]:
table = get_notification_table_to_use(service, notification_type, process_day, has_delete_task_run=False)
table = get_notification_table_to_use(service, notification_type, process_day, has_delete_task_run=False)
data_for_service_and_type = query_for_fact_status_data(
table=table,
start_date=start_date,
end_date=end_date,
notification_type=notification_type,
service_id=service.id
)
data_for_service_and_type = query_for_fact_status_data(
table=table,
start_date=start_date,
end_date=end_date,
notification_type=notification_type,
service_id=service.id
)
all_data_for_process_day += data_for_service_and_type
all_data_for_process_day += data_for_service_and_type
return all_data_for_process_day
@@ -92,10 +85,11 @@ def query_for_fact_status_data(table, start_date, end_date, notification_type, s
@transactional
def update_fact_notification_status(data, process_day):
def update_fact_notification_status(data, process_day, notification_type):
table = FactNotificationStatus.__table__
FactNotificationStatus.query.filter(
FactNotificationStatus.bst_date == process_day
FactNotificationStatus.bst_date == process_day,
FactNotificationStatus.notification_type == notification_type
).delete()
for row in data: