mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-21 07:51:13 -05:00
9f36b363f2
If you want to send a job on Monday morning, you should be able to schedule it on Friday. You shouldn’t need to work on the weekend. 96 hours is a full 4 days, so you can schedule a job at any time on Friday for any time on Monday. We’ve checked with the information assurance people, and they’re OK with us holding the data for this extra amount of time.
564 lines
17 KiB
Python
564 lines
17 KiB
Python
import re
|
|
from datetime import (
|
|
datetime,
|
|
date,
|
|
timedelta)
|
|
from flask_marshmallow.fields import fields
|
|
from marshmallow import (
|
|
post_load,
|
|
ValidationError,
|
|
validates,
|
|
validates_schema,
|
|
pre_load,
|
|
pre_dump,
|
|
post_dump
|
|
)
|
|
from marshmallow_sqlalchemy import field_for
|
|
|
|
from notifications_utils.recipients import (
|
|
validate_email_address,
|
|
InvalidEmailError,
|
|
validate_phone_number,
|
|
InvalidPhoneError,
|
|
validate_and_format_phone_number
|
|
)
|
|
|
|
from notifications_utils.renderers import PassThrough
|
|
|
|
from app import ma
|
|
from app import models
|
|
from app.dao.permissions_dao import permission_dao
|
|
|
|
|
|
def _validate_positive_number(value, msg="Not a positive integer"):
|
|
try:
|
|
page_int = int(value)
|
|
if page_int < 1:
|
|
raise ValidationError(msg)
|
|
except:
|
|
raise ValidationError(msg)
|
|
|
|
|
|
def _validate_datetime_not_more_than_96_hours_in_future(dte, msg="Date cannot be more than 96hrs in the future"):
|
|
if dte > datetime.utcnow() + timedelta(hours=96):
|
|
raise ValidationError(msg)
|
|
|
|
|
|
def _validate_not_in_future(dte, msg="Date cannot be in the future"):
|
|
if dte > date.today():
|
|
raise ValidationError(msg)
|
|
|
|
|
|
def _validate_not_in_past(dte, msg="Date cannot be in the past"):
|
|
if dte < date.today():
|
|
raise ValidationError(msg)
|
|
|
|
|
|
def _validate_datetime_not_in_future(dte, msg="Date cannot be in the future"):
|
|
if dte > datetime.utcnow():
|
|
raise ValidationError(msg)
|
|
|
|
|
|
def _validate_datetime_not_in_past(dte, msg="Date cannot be in the past"):
|
|
if dte < datetime.utcnow():
|
|
raise ValidationError(msg)
|
|
|
|
|
|
class BaseSchema(ma.ModelSchema):
|
|
|
|
def __init__(self, load_json=False, *args, **kwargs):
|
|
self.load_json = load_json
|
|
super(BaseSchema, self).__init__(*args, **kwargs)
|
|
|
|
@post_load
|
|
def make_instance(self, data):
|
|
"""Deserialize data to an instance of the model. Update an existing row
|
|
if specified in `self.instance` or loaded by primary key(s) in the data;
|
|
else create a new row.
|
|
:param data: Data to deserialize.
|
|
"""
|
|
if self.load_json:
|
|
return data
|
|
return super(BaseSchema, self).make_instance(data)
|
|
|
|
|
|
class UserSchema(BaseSchema):
|
|
|
|
permissions = fields.Method("user_permissions", dump_only=True)
|
|
password_changed_at = field_for(models.User, 'password_changed_at', format='%Y-%m-%d %H:%M:%S.%f')
|
|
created_at = field_for(models.User, 'created_at', format='%Y-%m-%d %H:%M:%S.%f')
|
|
|
|
def user_permissions(self, usr):
|
|
retval = {}
|
|
for x in permission_dao.get_permissions_by_user_id(usr.id):
|
|
service_id = str(x.service_id)
|
|
if service_id not in retval:
|
|
retval[service_id] = []
|
|
retval[service_id].append(x.permission)
|
|
return retval
|
|
|
|
class Meta:
|
|
model = models.User
|
|
exclude = (
|
|
"updated_at", "created_at", "user_to_service",
|
|
"_password", "verify_codes")
|
|
strict = True
|
|
|
|
|
|
class ProviderDetailsSchema(BaseSchema):
|
|
class Meta:
|
|
model = models.ProviderDetails
|
|
exclude = ("provider_rates", "provider_stats")
|
|
strict = True
|
|
|
|
|
|
class ServiceSchema(BaseSchema):
|
|
|
|
created_by = field_for(models.Service, 'created_by', required=True)
|
|
organisation = field_for(models.Service, 'organisation')
|
|
branding = field_for(models.Service, 'branding')
|
|
|
|
class Meta:
|
|
model = models.Service
|
|
exclude = ('updated_at',
|
|
'created_at',
|
|
'api_keys',
|
|
'templates',
|
|
'jobs',
|
|
'old_id',
|
|
'template_statistics',
|
|
'service_provider_stats',
|
|
'service_notification_stats')
|
|
strict = True
|
|
|
|
@validates('sms_sender')
|
|
def validate_sms_sender(self, value):
|
|
if value and not re.match(r'^[a-zA-Z0-9\s]+$', value):
|
|
raise ValidationError('Only alphanumeric characters allowed')
|
|
|
|
|
|
class DetailedServiceSchema(BaseSchema):
|
|
statistics = fields.Dict()
|
|
|
|
class Meta:
|
|
model = models.Service
|
|
exclude = (
|
|
'api_keys',
|
|
'templates',
|
|
'users',
|
|
'created_by',
|
|
'jobs',
|
|
'template_statistics',
|
|
'service_provider_stats',
|
|
'service_notification_stats',
|
|
'organisation'
|
|
)
|
|
|
|
|
|
class NotificationModelSchema(BaseSchema):
|
|
class Meta:
|
|
model = models.Notification
|
|
strict = True
|
|
exclude = ('_personalisation', 'job', 'service', 'template', 'api_key', '')
|
|
|
|
|
|
class BaseTemplateSchema(BaseSchema):
|
|
|
|
class Meta:
|
|
model = models.Template
|
|
exclude = ("service_id", "jobs")
|
|
strict = True
|
|
|
|
|
|
class TemplateSchema(BaseTemplateSchema):
|
|
|
|
created_by = field_for(models.Template, 'created_by', required=True)
|
|
|
|
@validates_schema
|
|
def validate_type(self, data):
|
|
template_type = data.get('template_type')
|
|
if template_type and template_type == 'email':
|
|
subject = data.get('subject')
|
|
if not subject or subject.strip() == '':
|
|
raise ValidationError('Invalid template subject', 'subject')
|
|
|
|
|
|
class TemplateHistorySchema(BaseSchema):
|
|
|
|
created_by = fields.Nested(UserSchema, only=['id', 'name', 'email_address'], dump_only=True)
|
|
created_at = field_for(models.Template, 'created_at', format='%Y-%m-%d %H:%M:%S.%f')
|
|
|
|
class Meta:
|
|
model = models.TemplateHistory
|
|
|
|
|
|
class NotificationsStatisticsSchema(BaseSchema):
|
|
class Meta:
|
|
model = models.NotificationStatistics
|
|
strict = True
|
|
|
|
@pre_dump
|
|
def handle_date_str(self, in_data):
|
|
if isinstance(in_data, dict) and 'day' in in_data:
|
|
in_data['day'] = datetime.strptime(in_data['day'], '%Y-%m-%d').date()
|
|
return in_data
|
|
|
|
|
|
class ApiKeySchema(BaseSchema):
|
|
|
|
created_by = field_for(models.ApiKey, 'created_by', required=True)
|
|
key_type = field_for(models.ApiKey, 'key_type', required=True)
|
|
|
|
class Meta:
|
|
model = models.ApiKey
|
|
exclude = ("service", "secret")
|
|
strict = True
|
|
|
|
|
|
class JobSchema(BaseSchema):
|
|
created_by_user = fields.Nested(UserSchema, attribute="created_by",
|
|
dump_to="created_by", only=["id", "name"], dump_only=True)
|
|
created_by = field_for(models.Job, 'created_by', required=True, load_only=True)
|
|
|
|
job_status = field_for(models.JobStatus, 'name', required=False)
|
|
|
|
scheduled_for = fields.DateTime()
|
|
|
|
@validates('scheduled_for')
|
|
def validate_scheduled_for(self, value):
|
|
_validate_datetime_not_in_past(value)
|
|
_validate_datetime_not_more_than_96_hours_in_future(value)
|
|
|
|
class Meta:
|
|
model = models.Job
|
|
exclude = (
|
|
'notifications',
|
|
'notifications_sent',
|
|
'notifications_delivered',
|
|
'notifications_failed')
|
|
strict = True
|
|
|
|
|
|
class RequestVerifyCodeSchema(ma.Schema):
|
|
|
|
class Meta:
|
|
strict = True
|
|
|
|
to = fields.Str(required=False)
|
|
|
|
|
|
class NotificationSchema(ma.Schema):
|
|
|
|
class Meta:
|
|
strict = True
|
|
|
|
personalisation = fields.Dict(required=False)
|
|
|
|
|
|
class SmsNotificationSchema(NotificationSchema):
|
|
to = fields.Str(required=True)
|
|
|
|
@validates('to')
|
|
def validate_to(self, value):
|
|
try:
|
|
validate_phone_number(value)
|
|
except InvalidPhoneError as error:
|
|
raise ValidationError('Invalid phone number: {}'.format(error))
|
|
|
|
@post_load
|
|
def format_phone_number(self, item):
|
|
item['to'] = validate_and_format_phone_number(item['to'])
|
|
return item
|
|
|
|
|
|
class EmailNotificationSchema(NotificationSchema):
|
|
to = fields.Str(required=True)
|
|
template = fields.Str(required=True)
|
|
|
|
@validates('to')
|
|
def validate_to(self, value):
|
|
try:
|
|
validate_email_address(value)
|
|
except InvalidEmailError as e:
|
|
raise ValidationError(e.message)
|
|
|
|
|
|
class SmsTemplateNotificationSchema(SmsNotificationSchema):
|
|
template = fields.Str(required=True)
|
|
job = fields.String()
|
|
|
|
|
|
class JobSmsTemplateNotificationSchema(SmsNotificationSchema):
|
|
template = fields.Str(required=True)
|
|
job = fields.String(required=True)
|
|
|
|
|
|
class JobEmailTemplateNotificationSchema(EmailNotificationSchema):
|
|
template = fields.Str(required=True)
|
|
job = fields.String(required=True)
|
|
|
|
|
|
class SmsAdminNotificationSchema(SmsNotificationSchema):
|
|
content = fields.Str(required=True)
|
|
|
|
|
|
class NotificationWithTemplateSchema(BaseSchema):
|
|
class Meta:
|
|
model = models.Notification
|
|
strict = True
|
|
exclude = ('_personalisation',)
|
|
|
|
template = fields.Nested(
|
|
TemplateSchema,
|
|
only=['id', 'version', 'name', 'template_type', 'content', 'subject'],
|
|
dump_only=True
|
|
)
|
|
job = fields.Nested(JobSchema, only=["id", "original_file_name"], dump_only=True)
|
|
personalisation = fields.Dict(required=False)
|
|
key_type = field_for(models.Notification, 'key_type', required=True)
|
|
key_name = fields.String()
|
|
|
|
@pre_dump
|
|
def add_api_key_name(self, in_data):
|
|
if in_data.api_key:
|
|
in_data.key_name = in_data.api_key.name
|
|
else:
|
|
in_data.key_name = None
|
|
return in_data
|
|
|
|
|
|
class NotificationWithPersonalisationSchema(NotificationWithTemplateSchema):
|
|
template_history = fields.Nested(TemplateHistorySchema,
|
|
only=['id', 'name', 'template_type', 'content', 'subject', 'version'],
|
|
dump_only=True)
|
|
|
|
class Meta(NotificationWithTemplateSchema.Meta):
|
|
# mark as many fields as possible as required since this is a public api.
|
|
# WARNING: Does _not_ reference fields computed in handle_template_merge, such as
|
|
# 'body', 'subject' [for emails], and 'content_char_count'
|
|
fields = (
|
|
# db rows
|
|
'id', 'to', 'job_row_number', 'template_version', 'billable_units', 'notification_type', 'created_at',
|
|
'sent_at', 'sent_by', 'updated_at', 'status', 'reference',
|
|
# computed fields
|
|
'personalisation',
|
|
# relationships
|
|
'service', 'job', 'api_key', 'template_history'
|
|
)
|
|
|
|
@pre_dump
|
|
def handle_personalisation_property(self, in_data):
|
|
self.personalisation = in_data.personalisation
|
|
return in_data
|
|
|
|
@post_dump
|
|
def handle_template_merge(self, in_data):
|
|
in_data['template'] = in_data.pop('template_history')
|
|
from notifications_utils.template import Template
|
|
template = Template(
|
|
in_data['template'],
|
|
in_data['personalisation'],
|
|
renderer=PassThrough()
|
|
)
|
|
in_data['body'] = template.replaced
|
|
template_type = in_data['template']['template_type']
|
|
if template_type == 'email':
|
|
in_data['subject'] = template.replaced_subject
|
|
in_data['content_char_count'] = None
|
|
else:
|
|
in_data['content_char_count'] = len(in_data['body'])
|
|
|
|
in_data.pop('personalisation', None)
|
|
in_data['template'].pop('content', None)
|
|
in_data['template'].pop('subject', None)
|
|
return in_data
|
|
|
|
|
|
class InvitedUserSchema(BaseSchema):
|
|
|
|
class Meta:
|
|
model = models.InvitedUser
|
|
strict = True
|
|
|
|
@validates('email_address')
|
|
def validate_to(self, value):
|
|
try:
|
|
validate_email_address(value)
|
|
except InvalidEmailError as e:
|
|
raise ValidationError(e.message)
|
|
|
|
|
|
class PermissionSchema(BaseSchema):
|
|
|
|
# Override generated fields
|
|
user = field_for(models.Permission, 'user', dump_only=True)
|
|
service = field_for(models.Permission, 'service', dump_only=True)
|
|
permission = field_for(models.Permission, 'permission')
|
|
|
|
__envelope__ = {
|
|
'single': 'permission',
|
|
'many': 'permissions',
|
|
}
|
|
|
|
class Meta:
|
|
model = models.Permission
|
|
exclude = ("created_at",)
|
|
strict = True
|
|
|
|
|
|
class EmailDataSchema(ma.Schema):
|
|
|
|
class Meta:
|
|
strict = True
|
|
|
|
email = fields.Str(required=True)
|
|
|
|
@validates('email')
|
|
def validate_email(self, value):
|
|
try:
|
|
validate_email_address(value)
|
|
except InvalidEmailError as e:
|
|
raise ValidationError(e.message)
|
|
|
|
|
|
class NotificationsFilterSchema(ma.Schema):
|
|
|
|
class Meta:
|
|
strict = True
|
|
|
|
template_type = fields.Nested(BaseTemplateSchema, only=['template_type'], many=True)
|
|
status = fields.Nested(NotificationModelSchema, only=['status'], many=True)
|
|
page = fields.Int(required=False)
|
|
page_size = fields.Int(required=False)
|
|
limit_days = fields.Int(required=False)
|
|
include_jobs = fields.Boolean(required=False)
|
|
include_from_test_key = fields.Boolean(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):
|
|
_validate_positive_number(value)
|
|
|
|
@validates('page_size')
|
|
def validate_page_size(self, value):
|
|
_validate_positive_number(value)
|
|
|
|
|
|
class TemplateStatisticsSchema(BaseSchema):
|
|
|
|
template = fields.Nested(TemplateSchema, only=["id", "name", "template_type"], dump_only=True)
|
|
|
|
class Meta:
|
|
model = models.TemplateStatistics
|
|
strict = True
|
|
|
|
|
|
class ServiceHistorySchema(ma.Schema):
|
|
id = fields.UUID()
|
|
name = fields.String()
|
|
created_at = fields.DateTime()
|
|
updated_at = fields.DateTime()
|
|
active = fields.Boolean()
|
|
message_limit = fields.Integer()
|
|
restricted = fields.Boolean()
|
|
email_from = fields.String()
|
|
created_by_id = fields.UUID()
|
|
version = fields.Integer()
|
|
|
|
|
|
class ApiKeyHistorySchema(ma.Schema):
|
|
id = fields.UUID()
|
|
name = fields.String()
|
|
service_id = fields.UUID()
|
|
expiry_date = fields.DateTime()
|
|
created_at = fields.DateTime()
|
|
updated_at = fields.DateTime()
|
|
created_by_id = fields.UUID()
|
|
|
|
|
|
class EventSchema(BaseSchema):
|
|
class Meta:
|
|
model = models.Event
|
|
strict = True
|
|
|
|
|
|
class OrganisationSchema(BaseSchema):
|
|
class Meta:
|
|
model = models.Organisation
|
|
strict = True
|
|
|
|
|
|
class DaySchema(ma.Schema):
|
|
|
|
class Meta:
|
|
strict = True
|
|
|
|
day = fields.Date(required=True)
|
|
|
|
@validates('day')
|
|
def validate_day(self, value):
|
|
_validate_not_in_future(value)
|
|
|
|
|
|
class UnarchivedTemplateSchema(BaseSchema):
|
|
archived = fields.Boolean(required=True)
|
|
|
|
@validates_schema
|
|
def validate_archived(self, data):
|
|
if data['archived']:
|
|
raise ValidationError('Template has been deleted', 'template')
|
|
|
|
|
|
user_schema = UserSchema()
|
|
user_schema_load_json = UserSchema(load_json=True)
|
|
service_schema = ServiceSchema()
|
|
service_schema_load_json = ServiceSchema(load_json=True)
|
|
detailed_service_schema = DetailedServiceSchema()
|
|
template_schema = TemplateSchema()
|
|
template_schema_load_json = TemplateSchema(load_json=True)
|
|
api_key_schema = ApiKeySchema()
|
|
api_key_schema_load_json = ApiKeySchema(load_json=True)
|
|
job_schema = JobSchema()
|
|
job_schema_load_json = JobSchema(load_json=True)
|
|
request_verify_code_schema = RequestVerifyCodeSchema()
|
|
sms_admin_notification_schema = SmsAdminNotificationSchema()
|
|
sms_template_notification_schema = SmsTemplateNotificationSchema()
|
|
job_sms_template_notification_schema = JobSmsTemplateNotificationSchema()
|
|
email_notification_schema = EmailNotificationSchema()
|
|
job_email_template_notification_schema = JobEmailTemplateNotificationSchema()
|
|
notification_schema = NotificationModelSchema()
|
|
notification_with_template_schema = NotificationWithTemplateSchema()
|
|
notification_with_personalisation_schema = NotificationWithPersonalisationSchema()
|
|
invited_user_schema = InvitedUserSchema()
|
|
permission_schema = PermissionSchema()
|
|
email_data_request_schema = EmailDataSchema()
|
|
notifications_statistics_schema = NotificationsStatisticsSchema()
|
|
notifications_filter_schema = NotificationsFilterSchema()
|
|
template_statistics_schema = TemplateStatisticsSchema()
|
|
service_history_schema = ServiceHistorySchema()
|
|
api_key_history_schema = ApiKeyHistorySchema()
|
|
template_history_schema = TemplateHistorySchema()
|
|
event_schema = EventSchema()
|
|
organisation_schema = OrganisationSchema()
|
|
provider_details_schema = ProviderDetailsSchema()
|
|
day_schema = DaySchema()
|
|
unarchived_template_schema = UnarchivedTemplateSchema()
|