Merge branch 'master' into integrate_MMG

Conflicts:
	app/notifications/rest.py
This commit is contained in:
Rebecca Law
2016-04-06 14:58:13 +01:00
21 changed files with 770 additions and 267 deletions

View File

@@ -59,6 +59,7 @@ def create_app(app_name=None):
from app.permission.rest import permission as permission_blueprint
from app.accept_invite.rest import accept_invite
from app.notifications_statistics.rest import notifications_statistics as notifications_statistics_blueprint
from app.template_statistics.rest import template_statistics as template_statistics_blueprint
application.register_blueprint(service_blueprint, url_prefix='/service')
application.register_blueprint(user_blueprint, url_prefix='/user')
@@ -70,6 +71,7 @@ def create_app(app_name=None):
application.register_blueprint(permission_blueprint, url_prefix='/permission')
application.register_blueprint(accept_invite, url_prefix='/invite')
application.register_blueprint(notifications_statistics_blueprint)
application.register_blueprint(template_statistics_blueprint)
return application

View File

@@ -1,7 +1,16 @@
from boto3 import resource
def get_job_from_s3(bucket_name, job_id):
def get_s3_job_object(bucket_name, job_id):
s3 = resource('s3')
key = s3.Object(bucket_name, '{}.csv'.format(job_id))
return key.get()['Body'].read().decode('utf-8')
return s3.Object(bucket_name, '{}.csv'.format(job_id))
def get_job_from_s3(bucket_name, job_id):
obj = get_s3_job_object(bucket_name, job_id)
return obj.get()['Body'].read().decode('utf-8')
def remove_job_from_s3(bucket_name, job_id):
obj = get_s3_job_object(bucket_name, job_id)
return obj.delete()

View File

@@ -1,3 +1,4 @@
import itertools
from datetime import datetime
from flask import current_app
@@ -12,7 +13,8 @@ from utils.template import Template
from utils.recipients import (
RecipientCSV,
validate_and_format_phone_number
validate_and_format_phone_number,
allowed_to_send_to
)
from app import (
@@ -33,8 +35,7 @@ from app.dao.invited_user_dao import delete_invitations_created_more_than_two_da
from app.dao.notifications_dao import (
dao_create_notification,
dao_update_notification,
delete_failed_notifications_created_more_than_a_week_ago,
delete_successful_notifications_created_more_than_a_day_ago,
delete_notifications_created_more_than_a_week_ago,
dao_get_notification_statistics_for_service_and_day,
update_notification_reference_by_id
)
@@ -50,11 +51,6 @@ from app.models import (
TEMPLATE_TYPE_SMS
)
from app.validation import (
allowed_send_to_email,
allowed_send_to_number
)
@notify_celery.task(name="delete-verify-codes")
def delete_verify_codes():
@@ -73,7 +69,7 @@ def delete_verify_codes():
def delete_successful_notifications():
try:
start = datetime.utcnow()
deleted = delete_successful_notifications_created_more_than_a_day_ago()
deleted = delete_notifications_created_more_than_a_week_ago('sent')
current_app.logger.info(
"Delete job started {} finished {} deleted {} successful notifications".format(
start,
@@ -90,7 +86,7 @@ def delete_successful_notifications():
def delete_failed_notifications():
try:
start = datetime.utcnow()
deleted = delete_failed_notifications_created_more_than_a_week_ago()
deleted = delete_notifications_created_more_than_a_week_ago('failed')
current_app.logger.info(
"Delete job started {} finished {} deleted {} failed notifications".format(
start,
@@ -185,29 +181,34 @@ def process_job(job_id):
job.processing_started = start
job.processing_finished = finished
dao_update_job(job)
remove_job.apply_async((str(job_id),), queue='remove-job')
current_app.logger.info(
"Job {} created at {} started at {} finished at {}".format(job_id, job.created_at, start, finished)
)
@notify_celery.task(name="remove-job")
def remove_job(job_id):
job = dao_get_job_by_id(job_id)
s3.remove_job_from_s3(job.bucket_name, job_id)
current_app.logger.info("Job {} has been removed from s3.".format(job_id))
@notify_celery.task(name="send-sms")
def send_sms(service_id, notification_id, encrypted_notification, created_at):
notification = encryption.decrypt(encrypted_notification)
service = dao_fetch_service_by_id(service_id)
client = firetext_client
restricted = False
if not service_allowed_to_send_to(notification['to'], service):
current_app.logger.info(
"SMS {} failed as restricted service".format(notification_id)
)
restricted = True
try:
status = 'sent'
can_send = True
if not allowed_send_to_number(service, notification['to']):
current_app.logger.info(
"SMS {} failed as restricted service".format(notification_id)
)
status = 'failed'
can_send = False
sent_at = datetime.utcnow()
notification_db_object = Notification(
id=notification_id,
@@ -215,7 +216,7 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
to=notification['to'],
service_id=service_id,
job_id=notification.get('job', None),
status=status,
status='failed' if restricted else 'sent',
created_at=datetime.strptime(created_at, DATETIME_FORMAT),
sent_at=sent_at,
sent_by=client.get_name()
@@ -223,30 +224,32 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
dao_create_notification(notification_db_object, TEMPLATE_TYPE_SMS)
if can_send:
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {}),
prefix=service.name
)
if restricted:
return
client.send_sms(
to=validate_and_format_phone_number(notification['to']),
content=template.replaced,
reference=str(notification_id)
)
except FiretextClientException as e:
current_app.logger.error(
"SMS notification {} failed".format(notification_id)
)
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"SMS {} created at {} sent at {}".format(notification_id, created_at, sent_at)
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {}),
prefix=service.name
)
client.send_sms(
to=validate_and_format_phone_number(notification['to']),
content=template.replaced,
reference=str(notification_id)
)
except FiretextClientException as e:
current_app.logger.error(
"SMS notification {} failed".format(notification_id)
)
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"SMS {} created at {} sent at {}".format(notification_id, created_at, sent_at)
)
except SQLAlchemyError as e:
current_app.logger.debug(e)
@@ -254,21 +257,18 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
@notify_celery.task(name="send-email")
def send_email(service_id, notification_id, subject, from_address, encrypted_notification, created_at):
notification = encryption.decrypt(encrypted_notification)
client = aws_ses_client
service = dao_fetch_service_by_id(service_id)
client = aws_ses_client
restricted = False
if not service_allowed_to_send_to(notification['to'], service):
current_app.logger.info(
"Email {} failed as restricted service".format(notification_id)
)
restricted = True
try:
status = 'sent'
can_send = True
if not allowed_send_to_email(service, notification['to']):
current_app.logger.info(
"Email {} failed as restricted service".format(notification_id)
)
status = 'failed'
can_send = False
sent_at = datetime.utcnow()
notification_db_object = Notification(
id=notification_id,
@@ -276,36 +276,38 @@ def send_email(service_id, notification_id, subject, from_address, encrypted_not
to=notification['to'],
service_id=service_id,
job_id=notification.get('job', None),
status=status,
status='failed' if restricted else 'sent',
created_at=datetime.strptime(created_at, DATETIME_FORMAT),
sent_at=sent_at,
sent_by=client.get_name()
)
dao_create_notification(notification_db_object, TEMPLATE_TYPE_EMAIL)
if can_send:
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {})
)
if restricted:
return
reference = client.send_email(
from_address,
notification['to'],
subject,
body=template.replaced,
html_body=template.as_HTML_email,
)
update_notification_reference_by_id(notification_id, reference)
except AwsSesClientException as e:
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"Email {} created at {} sent at {}".format(notification_id, created_at, sent_at)
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {})
)
reference = client.send_email(
from_address,
notification['to'],
subject,
body=template.replaced,
html_body=template.as_HTML_email,
)
update_notification_reference_by_id(notification_id, reference)
except AwsSesClientException as e:
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"Email {} created at {} sent at {}".format(notification_id, created_at, sent_at)
)
except SQLAlchemyError as e:
current_app.logger.debug(e)
@@ -423,3 +425,16 @@ def email_registration_verification(encrypted_verification_message):
url=verification_message['url']))
except AwsSesClientException as e:
current_app.logger.exception(e)
def service_allowed_to_send_to(recipient, service):
if not service.restricted:
return True
return allowed_to_send_to(
recipient,
itertools.chain.from_iterable(
[user.mobile_number, user.email_address] for user in service.users
)
)

View File

@@ -1,4 +1,8 @@
from sqlalchemy import desc
from sqlalchemy import (
desc,
func
)
from datetime import (
datetime,
timedelta,
@@ -6,6 +10,7 @@ from datetime import (
)
from flask import current_app
from werkzeug.datastructures import MultiDict
from app import db
from app.models import (
@@ -24,6 +29,23 @@ from app.clients import (
STATISTICS_REQUESTED
)
from functools import wraps
def transactional(func):
@wraps(func)
def commit_or_rollback(*args, **kwargs):
from flask import current_app
from app import db
try:
func(*args, **kwargs)
db.session.commit()
except Exception as e:
current_app.logger.error(e)
db.session.rollback()
raise
return commit_or_rollback
def dao_get_notification_statistics_for_service(service_id):
return NotificationStatistics.query.filter_by(
@@ -38,46 +60,56 @@ def dao_get_notification_statistics_for_service_and_day(service_id, day):
).order_by(desc(NotificationStatistics.day)).first()
def dao_get_template_statistics_for_service(service_id, limit_days=None):
filter = [TemplateStatistics.service_id == service_id]
if limit_days:
latest_stat = TemplateStatistics.query.filter_by(service_id=service_id).order_by(
desc(TemplateStatistics.day)).limit(1).first()
if latest_stat:
last_date_to_fetch = latest_stat.day - timedelta(days=limit_days)
else:
last_date_to_fetch = date.today() - timedelta(days=limit_days)
filter.append(TemplateStatistics.day > last_date_to_fetch)
return TemplateStatistics.query.filter(*filter).order_by(
desc(TemplateStatistics.day)).join(Template).order_by(func.lower(Template.name)).all()
@transactional
def dao_create_notification(notification, notification_type):
try:
if notification.job_id:
db.session.query(Job).filter_by(
id=notification.job_id
).update({
Job.notifications_sent: Job.notifications_sent + 1,
Job.updated_at: datetime.utcnow()
})
if notification.job_id:
db.session.query(Job).filter_by(
id=notification.job_id
).update({
Job.notifications_sent: Job.notifications_sent + 1,
Job.updated_at: datetime.utcnow()
})
update_count = db.session.query(NotificationStatistics).filter_by(
update_count = db.session.query(NotificationStatistics).filter_by(
day=notification.created_at.strftime('%Y-%m-%d'),
service_id=notification.service_id
).update(update_query(notification_type, 'requested'))
if update_count == 0:
stats = NotificationStatistics(
day=notification.created_at.strftime('%Y-%m-%d'),
service_id=notification.service_id
).update(update_query(notification_type, 'requested'))
if update_count == 0:
stats = NotificationStatistics(
day=notification.created_at.strftime('%Y-%m-%d'),
service_id=notification.service_id,
sms_requested=1 if notification_type == TEMPLATE_TYPE_SMS else 0,
emails_requested=1 if notification_type == TEMPLATE_TYPE_EMAIL else 0
)
db.session.add(stats)
update_count = db.session.query(TemplateStatistics).filter_by(
day=date.today(),
service_id=notification.service_id,
template_id=notification.template_id
).update({'usage_count': TemplateStatistics.usage_count + 1})
sms_requested=1 if notification_type == TEMPLATE_TYPE_SMS else 0,
emails_requested=1 if notification_type == TEMPLATE_TYPE_EMAIL else 0
)
db.session.add(stats)
if update_count == 0:
template_stats = TemplateStatistics(template_id=notification.template_id,
service_id=notification.service_id)
db.session.add(template_stats)
update_count = db.session.query(TemplateStatistics).filter_by(
day=date.today(),
service_id=notification.service_id,
template_id=notification.template_id
).update({'usage_count': TemplateStatistics.usage_count + 1})
db.session.add(notification)
db.session.commit()
except:
db.session.rollback()
raise
if update_count == 0:
template_stats = TemplateStatistics(template_id=notification.template_id,
service_id=notification.service_id)
db.session.add(template_stats)
db.session.add(notification)
def update_query(notification_type, status):
@@ -191,26 +223,32 @@ def get_notifications_for_service(service_id, filter_dict=None, page=1):
def filter_query(query, filter_dict=None):
if filter_dict and 'status' in filter_dict:
query = query.filter_by(status=filter_dict['status'])
if filter_dict and 'template_type' in filter_dict:
query = query.join(Template).filter(Template.template_type == filter_dict['template_type'])
if filter_dict is None:
filter_dict = MultiDict()
else:
filter_dict = MultiDict(filter_dict)
statuses = filter_dict.getlist('status') if 'status' in filter_dict else None
if statuses:
query = query.filter(Notification.status.in_(statuses))
template_types = filter_dict.getlist('template_type') if 'template_type' in filter_dict else None
if template_types:
query = query.join(Template).filter(Template.template_type.in_(template_types))
return query
def delete_successful_notifications_created_more_than_a_day_ago():
def delete_notifications_created_more_than_a_day_ago(status):
deleted = db.session.query(Notification).filter(
Notification.created_at < datetime.utcnow() - timedelta(days=1),
Notification.status == 'sent'
Notification.status == status
).delete()
db.session.commit()
return deleted
def delete_failed_notifications_created_more_than_a_week_ago():
def delete_notifications_created_more_than_a_week_ago(status):
deleted = db.session.query(Notification).filter(
Notification.created_at < datetime.utcnow() - timedelta(days=7),
Notification.status == 'failed'
Notification.status == status
).delete()
db.session.commit()
return deleted

View File

@@ -32,8 +32,10 @@ class PermissionDAO(DAOClass):
class Meta:
model = Permission
def get_query(self, filter_by_dict={}):
if isinstance(filter_by_dict, dict):
def get_query(self, filter_by_dict=None):
if filter_by_dict is None:
filter_by_dict = MultiDict()
else:
filter_by_dict = MultiDict(filter_by_dict)
query = self.Meta.model.query
if 'id' in filter_by_dict:

View File

@@ -37,6 +37,8 @@ def get_job_by_service_and_job_id(service_id, job_id):
def get_jobs_by_service(service_id):
jobs = dao_get_jobs_by_service_id(service_id)
data, errors = job_schema.dump(jobs, many=True)
if errors:
return jsonify(result="error", message=errors), 400
return jsonify(data=data)

View File

@@ -357,7 +357,7 @@ class TemplateStatistics(db.Model):
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, unique=False, nullable=False)
service = db.relationship('Service', backref=db.backref('template_statics', lazy='dynamic'))
service = db.relationship('Service', backref=db.backref('template_statistics', lazy='dynamic'))
template_id = db.Column(db.BigInteger, db.ForeignKey('templates.id'), index=True, nullable=False, unique=False)
template = db.relationship('Template')
usage_count = db.Column(db.BigInteger, index=False, unique=False, nullable=False, default=1)

View File

@@ -1,5 +1,6 @@
from datetime import datetime
import itertools
from flask import (
Blueprint,
jsonify,
@@ -8,7 +9,7 @@ from flask import (
url_for,
json
)
from utils.recipients import allowed_to_send_to, first_column_heading
from utils.template import Template
from app.clients.email.aws_ses import AwsSesResponses
from app import api_user, encryption, create_uuid, DATETIME_FORMAT, DATE_FORMAT
@@ -29,7 +30,6 @@ from app.schemas import (
notifications_filter_schema
)
from app.celery.tasks import send_sms, send_email
from app.validation import allowed_send_to_number, allowed_send_to_email
notifications = Blueprint('notifications', __name__)
@@ -327,12 +327,21 @@ def send_notification(notification_type):
}
), 400
if service.restricted and not allowed_to_send_to(
notification['to'],
itertools.chain.from_iterable(
[user.mobile_number, user.email_address] for user in service.users
)
):
return jsonify(
result="error", message={
'to': ['Invalid {} for restricted service'.format(first_column_heading[notification_type])]
}
), 400
notification_id = create_uuid()
if notification_type == 'sms':
if not allowed_send_to_number(service, notification['to']):
return jsonify(
result="error", message={'to': ['Invalid phone number for restricted service']}), 400
send_sms.apply_async((
service_id,
notification_id,
@@ -340,9 +349,6 @@ def send_notification(notification_type):
datetime.utcnow().strftime(DATETIME_FORMAT)
), queue='sms')
else:
if not allowed_send_to_email(service, notification['to']):
return jsonify(
result="error", message={'to': ['Email address not permitted for restricted service']}), 400
send_email.apply_async((
service_id,
notification_id,

View File

@@ -1,15 +1,27 @@
from flask_marshmallow.fields import fields
from . import ma
from . import models
from app.dao.permissions_dao import permission_dao
from marshmallow import (post_load, ValidationError, validates, validates_schema)
from marshmallow import (
post_load,
ValidationError,
validates,
validates_schema,
pre_load
)
from marshmallow_sqlalchemy import field_for
from utils.recipients import (
validate_email_address, InvalidEmailError,
validate_phone_number, InvalidPhoneError,
validate_email_address,
InvalidEmailError,
validate_phone_number,
InvalidPhoneError,
validate_and_format_phone_number
)
from app import ma
from app import models
from app.dao.permissions_dao import permission_dao
# TODO I think marshmallow provides a better integration and error handling.
# Would be better to replace functionality in dao with the marshmallow supported
@@ -19,7 +31,7 @@ from utils.recipients import (
class BaseSchema(ma.ModelSchema):
def __init__(self, *args, load_json=False, **kwargs):
def __init__(self, load_json=False, *args, **kwargs):
self.load_json = load_json
super(BaseSchema, self).__init__(*args, **kwargs)
@@ -74,6 +86,11 @@ class ServiceSchema(BaseSchema):
exclude = ("updated_at", "created_at", "api_keys", "templates", "jobs", 'old_id')
class NotificationModelSchema(BaseSchema):
class Meta:
model = models.Notification
class TemplateSchema(BaseSchema):
class Meta:
model = models.Template
@@ -203,10 +220,29 @@ class EmailDataSchema(ma.Schema):
class NotificationsFilterSchema(ma.Schema):
template_type = field_for(models.Template, 'template_type', load_only=True, required=False)
status = field_for(models.Notification, 'status', load_only=True, required=False)
template_type = fields.Nested(TemplateSchema, only='template_type', many=True)
status = fields.Nested(NotificationModelSchema, only='status', many=True)
page = fields.Int(required=False)
@pre_load
def handle_multidict(self, in_data):
if isinstance(in_data, dict) and hasattr(in_data, 'getlist'):
out_data = dict([(k, in_data.get(k)) for k in in_data.keys()])
if 'template_type' in in_data:
out_data['template_type'] = [{'template_type': x} for x in in_data.getlist('template_type')]
if 'status' in in_data:
out_data['status'] = [{"status": x} for x in in_data.getlist('status')]
return out_data
@post_load
def convert_schema_object_to_field(self, in_data):
if 'template_type' in in_data:
in_data['template_type'] = [x.template_type for x in in_data['template_type']]
if 'status' in in_data:
in_data['status'] = [x.status for x in in_data['status']]
return in_data
@validates('page')
def validate_page(self, value):
try:
@@ -216,6 +252,15 @@ class NotificationsFilterSchema(ma.Schema):
except:
raise ValidationError("Not a positive integer")
class TemplateStatisticsSchema(BaseSchema):
template = fields.Nested(TemplateSchema, only=["id", "name", "template_type"], dump_only=True)
class Meta:
model = models.TemplateStatistics
user_schema = UserSchema()
user_schema_load_json = UserSchema(load_json=True)
service_schema = ServiceSchema()
@@ -239,3 +284,4 @@ permission_schema = PermissionSchema()
email_data_request_schema = EmailDataSchema()
notifications_statistics_schema = NotificationsStatisticsSchema()
notifications_filter_schema = NotificationsFilterSchema()
template_statistics_schema = TemplateStatisticsSchema()

View File

View File

@@ -0,0 +1,36 @@
from flask import (
Blueprint,
jsonify,
request,
current_app
)
from app.dao.notifications_dao import dao_get_template_statistics_for_service
from app.schemas import template_statistics_schema
template_statistics = Blueprint('template-statistics',
__name__,
url_prefix='/service/<service_id>/template-statistics')
from app.errors import register_errors
register_errors(template_statistics)
@template_statistics.route('')
def get_template_statistics_for_service(service_id):
if request.args.get('limit_days'):
try:
limit_days = int(request.args['limit_days'])
except ValueError as e:
error = '{} is not an integer'.format(request.args['limit_days'])
current_app.logger.error(error)
return jsonify(result="error", message={'limit_days': [error]}), 400
else:
limit_days = None
stats = dao_get_template_statistics_for_service(service_id, limit_days=limit_days)
data, errors = template_statistics_schema.dump(stats, many=True)
if errors:
return jsonify(result="error", message=errors), 400
return jsonify(data=data)

View File

@@ -1,15 +0,0 @@
from utils.recipients import format_phone_number, validate_phone_number
def allowed_send_to_number(service, to):
if service.restricted and format_phone_number(validate_phone_number(to)) not in [
format_phone_number(validate_phone_number(user.mobile_number)) for user in service.users
]:
return False
return True
def allowed_send_to_email(service, to):
if service.restricted and to not in [user.email_address for user in service.users]:
return False
return True