mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-22 16:31:15 -05:00
800afc9e44
This is checked on 3rd party API calls, but jobs (CSV files) were able expected to only allow valid files. Change in tack means we want to have restricted notification failures reported in the UI.
321 lines
11 KiB
Python
321 lines
11 KiB
Python
import uuid
|
|
|
|
from sqlalchemy import UniqueConstraint, Sequence
|
|
from . import db
|
|
import datetime
|
|
from sqlalchemy.dialects.postgresql import (UUID, ARRAY)
|
|
from app.encryption import (
|
|
hashpw,
|
|
check_hash
|
|
)
|
|
|
|
|
|
def filter_null_value_fields(obj):
|
|
return dict(
|
|
filter(lambda x: x[1] is not None, obj.items())
|
|
)
|
|
|
|
|
|
class User(db.Model):
|
|
__tablename__ = 'users'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String, nullable=False, index=True, unique=False)
|
|
email_address = db.Column(db.String(255), nullable=False, index=True, unique=True)
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False,
|
|
default=datetime.datetime.now)
|
|
updated_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True,
|
|
onupdate=datetime.datetime.now)
|
|
_password = db.Column(db.String, index=False, unique=False, nullable=False)
|
|
mobile_number = db.Column(db.String, index=False, unique=False, nullable=False)
|
|
password_changed_at = db.Column(db.DateTime, index=False, unique=False, nullable=True)
|
|
logged_in_at = db.Column(db.DateTime, nullable=True)
|
|
failed_login_count = db.Column(db.Integer, nullable=False, default=0)
|
|
state = db.Column(db.String, nullable=False, default='pending')
|
|
|
|
@property
|
|
def password(self):
|
|
raise AttributeError("Password not readable")
|
|
|
|
@password.setter
|
|
def password(self, password):
|
|
self._password = hashpw(password)
|
|
|
|
def check_password(self, password):
|
|
return check_hash(password, self._password)
|
|
|
|
|
|
user_to_service = db.Table(
|
|
'user_to_service',
|
|
db.Model.metadata,
|
|
db.Column('user_id', db.Integer, db.ForeignKey('users.id')),
|
|
db.Column('service_id', UUID(as_uuid=True), db.ForeignKey('services.id')),
|
|
|
|
UniqueConstraint('user_id', 'service_id', name='uix_user_to_service')
|
|
|
|
)
|
|
|
|
|
|
class Service(db.Model):
|
|
__tablename__ = 'services'
|
|
|
|
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
old_id = db.Column(db.Integer, Sequence('services_id_seq'), nullable=False)
|
|
name = db.Column(db.String(255), nullable=False, unique=True)
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False,
|
|
default=datetime.datetime.now)
|
|
updated_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True,
|
|
onupdate=datetime.datetime.now)
|
|
active = db.Column(db.Boolean, index=False, unique=False, nullable=False)
|
|
limit = db.Column(db.BigInteger, index=False, unique=False, nullable=False)
|
|
users = db.relationship(
|
|
'User',
|
|
secondary=user_to_service,
|
|
backref=db.backref('user_to_service', lazy='dynamic'))
|
|
restricted = db.Column(db.Boolean, index=False, unique=False, nullable=False)
|
|
email_from = db.Column(db.Text, index=False, unique=True, nullable=False)
|
|
|
|
|
|
class ApiKey(db.Model):
|
|
__tablename__ = 'api_key'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(255), nullable=False)
|
|
secret = db.Column(db.String(255), unique=True, nullable=False)
|
|
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, nullable=False)
|
|
service = db.relationship('Service', backref=db.backref('api_keys', lazy='dynamic'))
|
|
expiry_date = db.Column(db.DateTime)
|
|
|
|
__table_args__ = (
|
|
UniqueConstraint('service_id', 'name', name='uix_service_to_key_name'),
|
|
)
|
|
|
|
|
|
TEMPLATE_TYPES = ['sms', 'email', 'letter']
|
|
|
|
|
|
class Template(db.Model):
|
|
__tablename__ = 'templates'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(255), nullable=False)
|
|
template_type = db.Column(db.Enum(*TEMPLATE_TYPES, name='template_type'), nullable=False)
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False,
|
|
default=datetime.datetime.now)
|
|
updated_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True,
|
|
onupdate=datetime.datetime.now)
|
|
content = db.Column(db.Text, index=False, unique=False, nullable=False)
|
|
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('templates', lazy='dynamic'))
|
|
subject = db.Column(db.Text, index=False, unique=True, nullable=True)
|
|
|
|
|
|
JOB_STATUS_TYPES = ['pending', 'in progress', 'finished']
|
|
|
|
|
|
class Job(db.Model):
|
|
__tablename__ = 'jobs'
|
|
|
|
id = db.Column(UUID(as_uuid=True), primary_key=True)
|
|
original_file_name = db.Column(db.String, nullable=False)
|
|
bucket_name = db.Column(db.String, nullable=False)
|
|
file_name = db.Column(db.String, nullable=False)
|
|
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('jobs', lazy='dynamic'))
|
|
template_id = db.Column(db.BigInteger, db.ForeignKey('templates.id'), index=True, unique=False)
|
|
template = db.relationship('Template', backref=db.backref('jobs', lazy='dynamic'))
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False,
|
|
default=datetime.datetime.now)
|
|
updated_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True,
|
|
onupdate=datetime.datetime.now)
|
|
status = db.Column(db.Enum(*JOB_STATUS_TYPES, name='job_status_types'), nullable=False, default='pending')
|
|
notification_count = db.Column(db.Integer, nullable=False)
|
|
processing_started = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True)
|
|
processing_finished = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True)
|
|
|
|
|
|
VERIFY_CODE_TYPES = ['email', 'sms']
|
|
|
|
|
|
class VerifyCode(db.Model):
|
|
__tablename__ = 'verify_codes'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), index=True, nullable=False)
|
|
user = db.relationship('User', backref=db.backref('verify_codes', lazy='dynamic'))
|
|
_code = db.Column(db.String, nullable=False)
|
|
code_type = db.Column(db.Enum(*VERIFY_CODE_TYPES, name='verify_code_types'),
|
|
index=False, unique=False, nullable=False)
|
|
expiry_datetime = db.Column(db.DateTime, nullable=False)
|
|
code_used = db.Column(db.Boolean, default=False)
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False,
|
|
default=datetime.datetime.now)
|
|
|
|
@property
|
|
def code(self):
|
|
raise AttributeError("Code not readable")
|
|
|
|
@code.setter
|
|
def code(self, cde):
|
|
self._code = hashpw(cde)
|
|
|
|
def check_code(self, cde):
|
|
return check_hash(cde, self._code)
|
|
|
|
|
|
NOTIFICATION_STATUS_TYPES = ['sent', 'failed', 'failed - restricted service']
|
|
|
|
|
|
class Notification(db.Model):
|
|
|
|
__tablename__ = 'notifications'
|
|
|
|
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
to = db.Column(db.String, nullable=False)
|
|
job_id = db.Column(UUID(as_uuid=True), db.ForeignKey('jobs.id'), index=True, unique=False)
|
|
job = db.relationship('Job', backref=db.backref('notifications', lazy='dynamic'))
|
|
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, unique=False)
|
|
service = db.relationship('Service')
|
|
template_id = db.Column(db.BigInteger, db.ForeignKey('templates.id'), index=True, unique=False)
|
|
template = db.relationship('Template')
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False)
|
|
sent_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True)
|
|
sent_by = db.Column(db.String, nullable=True)
|
|
updated_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=True,
|
|
onupdate=datetime.datetime.now)
|
|
status = db.Column(
|
|
db.Enum(*NOTIFICATION_STATUS_TYPES, name='notification_status_types'), nullable=False, default='sent')
|
|
|
|
|
|
INVITED_USER_STATUS_TYPES = ['pending', 'accepted', 'cancelled']
|
|
|
|
|
|
class InvitedUser(db.Model):
|
|
|
|
__tablename__ = 'invited_users'
|
|
|
|
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
email_address = db.Column(db.String(255), nullable=False)
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), index=True, nullable=False)
|
|
from_user = db.relationship('User')
|
|
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, unique=False)
|
|
service = db.relationship('Service')
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False,
|
|
default=datetime.datetime.now)
|
|
status = db.Column(
|
|
db.Enum(*INVITED_USER_STATUS_TYPES, name='invited_users_status_types'), nullable=False, default='pending')
|
|
permissions = db.Column(db.String, nullable=False)
|
|
|
|
# would like to have used properties for this but haven't found a way to make them
|
|
# play nice with marshmallow yet
|
|
def get_permissions(self):
|
|
return self.permissions.split(',')
|
|
|
|
|
|
# Service Permissions
|
|
MANAGE_USERS = 'manage_users'
|
|
MANAGE_TEMPLATES = 'manage_templates'
|
|
MANAGE_SETTINGS = 'manage_settings'
|
|
SEND_TEXTS = 'send_texts'
|
|
SEND_EMAILS = 'send_emails'
|
|
SEND_LETTERS = 'send_letters'
|
|
MANAGE_API_KEYS = 'manage_api_keys'
|
|
ACCESS_DEVELOPER_DOCS = 'access_developer_docs'
|
|
|
|
# List of permissions
|
|
PERMISSION_LIST = [
|
|
MANAGE_USERS,
|
|
MANAGE_TEMPLATES,
|
|
MANAGE_SETTINGS,
|
|
SEND_TEXTS,
|
|
SEND_EMAILS,
|
|
SEND_LETTERS,
|
|
MANAGE_API_KEYS,
|
|
ACCESS_DEVELOPER_DOCS]
|
|
|
|
|
|
class Permission(db.Model):
|
|
__tablename__ = 'permissions'
|
|
|
|
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
# Service id is optional, if the service is omitted we will assume the permission is not service specific.
|
|
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, unique=False, nullable=True)
|
|
service = db.relationship('Service')
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), index=True, nullable=False)
|
|
user = db.relationship('User')
|
|
permission = db.Column(
|
|
db.Enum(*PERMISSION_LIST, name='permission_types'),
|
|
index=False,
|
|
unique=False,
|
|
nullable=False)
|
|
created_at = db.Column(
|
|
db.DateTime,
|
|
index=False,
|
|
unique=False,
|
|
nullable=False,
|
|
default=datetime.datetime.now)
|
|
|
|
__table_args__ = (
|
|
UniqueConstraint('service_id', 'user_id', 'permission', name='uix_service_user_permission'),
|
|
)
|