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

Merge branch 'master' into seperate-queues-for-sending-and-for-db

Browse Source
This commit is contained in:
Martyn Inglis
2016-08-30 14:25:14 +01:00
parent 354efbdd2c 4a078e3c61
commit f3c614634f
31 changed files with 490 additions and 844 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
app/celery/provider_tasks.py
View File

@@ -13,7 +13,6 @@ from app import notify_celery, statsd_client, clients, create_uuid
from app.clients.email import EmailClientException
from app.clients.sms import SmsClientException
from app.dao.notifications_dao import (
update_provider_stats,
get_notification_by_id,
dao_update_notification,
update_notification_status_by_id
@@ -78,13 +77,6 @@ def send_sms_to_provider(self, service_id, notification_id):
)
notification.billable_units = get_sms_fragment_count(template.replaced_content_count)
update_provider_stats(
notification_id,
SMS_TYPE,
provider.get_name(),
billable_units=notification.billable_units
)
notification.sent_at = datetime.utcnow()
notification.sent_by = provider.get_name()
notification.status = 'sending'
@@ -101,7 +93,7 @@ def send_sms_to_provider(self, service_id, notification_id):
"RETRY FAILED: task send_sms_to_provider failed for notification {}".format(notification.id),
e
)
update_notification_status_by_id(notification.id, 'technical-failure', 'failure')
update_notification_status_by_id(notification.id, 'technical-failure')
current_app.logger.info(
"SMS {} sent to provider at {}".format(notification_id, notification.sent_at)
@@ -113,7 +105,7 @@ def send_sms_to_provider(self, service_id, notification_id):
def provider_to_use(notification_type, notification_id):
active_providers_in_order = [
provider for provider in get_provider_details_by_notification_type(notification_type) if provider.active
]
]
if not active_providers_in_order:
current_app.logger.error(
@@ -163,12 +155,6 @@ def send_email_to_provider(self, service_id, notification_id):
reply_to_address=service.reply_to_email_address,
)
update_provider_stats(
notification_id,
EMAIL_TYPE,
provider.get_name(),
billable_units=1
)
notification.reference = reference
notification.sent_at = datetime.utcnow()
notification.sent_by = provider.get_name(),
@@ -186,7 +172,7 @@ def send_email_to_provider(self, service_id, notification_id):
"RETRY FAILED: task send_email_to_provider failed for notification {}".format(notification.id),
e
)
update_notification_status_by_id(notification.id, 'technical-failure', 'failure')
update_notification_status_by_id(notification.id, 'technical-failure')
current_app.logger.info(
"Email {} sent to provider at {}".format(notification_id, notification.sent_at)

50
app/celery/scheduled_tasks.py
View File

@@ -6,10 +6,30 @@ from sqlalchemy.exc import SQLAlchemyError
from app import notify_celery
from app.clients import STATISTICS_FAILURE
from app.dao.invited_user_dao import delete_invitations_created_more_than_two_days_ago
from app.dao.jobs_dao import dao_get_scheduled_jobs, dao_update_job
from app.dao.notifications_dao import delete_notifications_created_more_than_a_week_ago, get_notifications, \
update_notification_status_by_id
from app.dao.users_dao import delete_codes_older_created_more_than_a_day_ago
from app.statsd_decorators import statsd
from app.models import JOB_STATUS_PENDING
from app.celery.tasks import process_job
@notify_celery.task(name="run-scheduled-jobs")
@statsd(namespace="tasks")
def run_scheduled_jobs():
try:
jobs = dao_get_scheduled_jobs()
for job in jobs:
job.job_status = JOB_STATUS_PENDING
dao_update_job(job)
process_job.apply_async([str(job.id)], queue="process-job")
current_app.logger.info(
"Job ID {} added to process job queue".format(job.id)
)
except SQLAlchemyError as e:
current_app.logger.exception("Failed to run scheduled jobs", e)
raise
@notify_celery.task(name="delete-verify-codes")
@@ -21,8 +41,8 @@ def delete_verify_codes():
current_app.logger.info(
"Delete job started {} finished {} deleted {} verify codes".format(start, datetime.utcnow(), deleted)
)
except SQLAlchemyError:
current_app.logger.info("Failed to delete verify codes")
except SQLAlchemyError as e:
current_app.logger.exception("Failed to delete verify codes", e)
raise
@@ -39,8 +59,8 @@ def delete_successful_notifications():
deleted
)
)
except SQLAlchemyError:
current_app.logger.info("Failed to delete successful notifications")
except SQLAlchemyError as e:
current_app.logger.exception("Failed to delete successful notifications", e)
raise
@@ -60,8 +80,8 @@ def delete_failed_notifications():
deleted
)
)
except SQLAlchemyError:
current_app.logger.info("Failed to delete failed notifications")
except SQLAlchemyError as e:
current_app.logger.exception("Failed to delete failed notifications", e)
raise
@@ -74,8 +94,8 @@ def delete_invitations():
current_app.logger.info(
"Delete job started {} finished {} deleted {} invitations".format(start, datetime.utcnow(), deleted)
)
except SQLAlchemyError:
current_app.logger.info("Failed to delete invitations")
except SQLAlchemyError as e:
current_app.logger.exception("Failed to delete invitations", e)
raise
@@ -88,15 +108,15 @@ def timeout_notifications():
for noti in notifications:
try:
if (now - noti.created_at) > timedelta(
seconds=current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD')
seconds=current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD')
):
# TODO: think about making this a bulk update rather than one at a time.
updated = update_notification_status_by_id(noti.id, 'temporary-failure', STATISTICS_FAILURE)
updated = update_notification_status_by_id(noti.id, 'temporary-failure')
if updated:
current_app.logger.info(("Timeout period reached for notification ({})"
", status has been updated.").format(noti.id))
current_app.logger.info(
"Timeout period reached for notification ({}), status has been updated.".format(noti.id))
except Exception as e:
current_app.logger.exception(e)
current_app.logger.error((
"Exception raised trying to timeout notification ({})"
", skipping notification update.").format(noti.id))
current_app.logger.error(
"Exception raised trying to timeout notification ({}) skipping notification update.".format(noti.id)
)

22
app/celery/tasks.py
View File

@@ -2,7 +2,6 @@ import itertools
from datetime import (datetime)
from flask import current_app
from monotonic import monotonic
from notifications_utils.recipients import (
RecipientCSV,
allowed_to_send_to
@@ -13,7 +12,6 @@ from sqlalchemy.exc import SQLAlchemyError
from app import (
create_uuid,
DATETIME_FORMAT,
DATE_FORMAT,
notify_celery,
encryption
)
@@ -23,11 +21,8 @@ from app.dao.jobs_dao import (
dao_update_job,
dao_get_job_by_id
)
from app.dao.notifications_dao import (
dao_create_notification,
dao_get_notification_statistics_for_service_and_day
)
from app.dao.services_dao import dao_fetch_service_by_id
from app.dao.notifications_dao import (dao_create_notification)
from app.dao.services_dao import dao_fetch_service_by_id, dao_fetch_todays_stats_for_service
from app.dao.templates_dao import dao_get_template_by_id
from app.models import (
Notification,
@@ -46,14 +41,7 @@ def process_job(job_id):
service = job.service
stats = dao_get_notification_statistics_for_service_and_day(
service_id=service.id,
day=job.created_at.strftime(DATE_FORMAT)
)
total_sent = 0
if stats:
total_sent = stats.emails_requested + stats.sms_requested
total_sent = sum(row.count for row in dao_fetch_todays_stats_for_service(service.id))
if total_sent + job.notification_count > service.message_limit:
job.status = 'sending limits exceeded'
@@ -87,7 +75,7 @@ def process_job(job_id):
'personalisation': {
key: personalisation.get(key)
for key in template.placeholders
}
}
})
if template.template_type == SMS_TYPE:
@@ -212,7 +200,7 @@ def _save_notification(created_at, notification, notification_id, service_id, no
api_key_id=api_key_id,
key_type=key_type
)
dao_create_notification(notification_db_object, notification_type)
dao_create_notification(notification_db_object)
def service_allowed_to_send_to(recipient, service):