Split out the Users model & give it responsibility

api/app/__init__.py

@@ -1,14 +1,11 @@
 from flask import Flask
-
 from app.logger import logger
-from app.create_sa import create_sa
+from flask_bcrypt import Bcrypt
 
 app = Flask(__name__)
+bcrypt = Bcrypt(app)
 
 
 def create_application():
-
     app.debug = True
-    create_sa()
-
     return app

api/application.py

@@ -1,18 +1,23 @@
 from waitress import serve
-
 from app import create_application
 from app.graphql_endpoint import add_graphql_endpoint
-
 from db import db_session
+from app.create_sa import create_sa
 
 app = create_application()
 add_graphql_endpoint(app)
 
 
+@app.before_first_request
+def setup_super_admin():
+    print("Creating Super Admin User")
+    create_sa()
+
+
 @app.teardown_request
 def shutdown_session(exception=None):
     db_session.remove()
 
 
-if __name__ == '__main__':
-    serve(app, host='0.0.0.0', port=5000)
+if __name__ == "__main__":
+    serve(app, host="0.0.0.0", port=5000)

api/functions/create_user.py

@@ -1,5 +1,4 @@
 from graphql import GraphQLError
-from flask_bcrypt import Bcrypt
 from flask import current_app as app
 
 from functions.input_validators import *
@@ -30,17 +29,14 @@ def create_user(display_name, password, confirm_password, user_name):
     if password != confirm_password:
         raise GraphQLError(error_passwords_do_not_match())
 
-    bcrypt = Bcrypt(app)  # Create the bcrypt object that handles password hashing and verifying.
-
-    user = User.query.filter(User.user_name == user_name).first()
+    user = User.find_by_user_name(user_name)
 
     if user is None:
         user = User(
             user_name=user_name,
             display_name=display_name,
-            preferred_lang='English',
-            user_password=bcrypt.generate_password_hash(
-                password=password).decode('UTF-8')  # Hash the password
+            preferred_lang="English",
+            user_password=password,
         )
 
         db_session.add(user)

api/functions/sign_in_user.py

@@ -2,14 +2,12 @@
 import datetime
 import os
 from graphql import GraphQLError
-from flask_bcrypt import Bcrypt
+from app import bcrypt
 from flask import current_app as app
 from sqlalchemy.orm import load_only
-
 from functions.input_validators import *
 from functions.error_messages import *
 from functions.orm_to_dict import orm_to_dict
-
 from models import Users
 from models import User_affiliations
 from db import db_session
@@ -24,63 +22,61 @@ def sign_in_user(user_name, password):
     """
     user_name = cleanse_input(user_name)
     password = cleanse_input(password)
-    user = Users.query.filter(Users.user_name == user_name).first()
+    user = Users.find_by_user_name(user_name)
 
     if user is None:
         raise GraphQLError(error_user_does_not_exist())
 
     # Checks the amount of failed login attempts and if the time since the last
     # Attempt was more than 30min (1800s)
-    if user.failed_login_attempts and (user.failed_login_attempt_time + 1800) \
-            < datetime.datetime.now().timestamp():
+    if (
+        user.failed_login_attempts
+        and (user.failed_login_attempt_time + 1800)
+        < datetime.datetime.now().timestamp()
+    ):
         raise GraphQLError(error_too_many_failed_login_attempts())
 
-    bcrypt = Bcrypt(app)  # Create the bcrypt object that will handle password hashing and verification
-
     email_match = user_name == user.user_name
     password_match = bcrypt.check_password_hash(user.user_password, password)
 
     # If the given user credentials are valid
     if email_match and password_match:
         # Fetch user's role from the database and include it as claims on the JWT being generated
-        user_aff = User_affiliations.query.filter(User_affiliations.user_id == user.id).all()
+        user_aff = User_affiliations.query.filter(
+            User_affiliations.user_id == user.id
+        ).all()
         user_aff = orm_to_dict(user_aff)
         if len(user_aff):
             user_roles = []
             for select in user_aff:
                 temp_dict = {
-                    'user_id': select['user_id'],
-                    'org_id': select['organization_id'],
-                    'permission': select['permission']
+                    "user_id": select["user_id"],
+                    "org_id": select["organization_id"],
+                    "permission": select["permission"],
                 }
                 user_roles.append(temp_dict)
         else:
-            user_roles = ['none']
+            user_roles = ["none"]
         try:
             payload = {
-                'exp': datetime.datetime.utcnow() + datetime.timedelta(days=0, seconds=1800),
-                'iat': datetime.datetime.utcnow(),
-                'user_id': user.id,
-                "roles": user_roles
+                "exp": datetime.datetime.utcnow()
+                + datetime.timedelta(days=0, seconds=1800),
+                "iat": datetime.datetime.utcnow(),
+                "user_id": user.id,
+                "roles": user_roles,
             }
             token = jwt.encode(
-                payload,
-                str(os.getenv('SUPER_SECRET_SALT')),
-                algorithm='HS256'
-            ).decode('utf-8')
+                payload, str(os.getenv("SUPER_SECRET_SALT")), algorithm="HS256"
+            ).decode("utf-8")
         except Exception as e:
-            raise GraphQLError('Token Generation Error: ' + str(e))
+            raise GraphQLError("Token Generation Error: " + str(e))
 
         # A temporary dictionary that will be returned to the graphql resolver
-        temp_dict = {
-            'auth_token': token,
-            'user': user
-        }
+        temp_dict = {"auth_token": token, "user": user}
 
-        Users.query.filter(Users.user_name == user_name).update({
-            'failed_login_attempts': 0,
-            'failed_login_attempt_time': 0
-        })
+        Users.query.filter(Users.user_name == user_name).update(
+            {"failed_login_attempts": 0, "failed_login_attempt_time": 0}
+        )
         db_session.commit()
 
         return temp_dict
@@ -89,11 +85,12 @@ def sign_in_user(user_name, password):
         # Generate a timestamp and also add that to the user update.
         time_stamp = datetime.datetime.now().timestamp()
 
-        Users.query.filter(Users.user_name == user_name)\
-            .update({
-                'failed_login_attempts': Users.failed_login_attempts + 1,
-                'failed_login_attempt_time': time_stamp
-             })
+        Users.query.filter(Users.user_name == user_name).update(
+            {
+                "failed_login_attempts": Users.failed_login_attempts + 1,
+                "failed_login_attempt_time": time_stamp,
+            }
+        )
         db_session.commit()
 
         raise GraphQLError(error_invalid_credentials())

api/functions/update_user_password.py

@@ -1,11 +1,8 @@
 import datetime
-
 from graphql import GraphQLError
-from flask_bcrypt import Bcrypt
-
+from app import bcrypt
 from functions.input_validators import *
 from functions.error_messages import *
-
 from models import Users as User
 from app import app
 from db import db_session
@@ -29,25 +26,24 @@ def update_password(user_name, password, confirm_password):
     if password != confirm_password:
         raise GraphQLError(error_passwords_do_not_match())
 
-    user = User.query.filter(User.user_name == user_name).first()
+    user = User.find_by_user_name(user_name)
 
     if user is None:
         raise GraphQLError(error_user_does_not_exist())
 
-    bcrypt = Bcrypt(app)  # Create the bcrypt object that will handle password hashing and verification
-
-    user = User.query.filter(User.user_name == user_name) \
-        .update({
-            'user_password': bcrypt.generate_password_hash(password).decode('UTF-8'),
-            'failed_login_attempts': 0,
-            'failed_login_attempt_time': 0
-        })
+    user = User.query.filter(User.user_name == user_name).update(
+        {
+            "user_password": bcrypt.generate_password_hash(password).decode("UTF-8"),
+            "failed_login_attempts": 0,
+            "failed_login_attempt_time": 0,
+        }
+    )
 
     db_session.commit()
 
     if not user:
         raise GraphQLError(error_password_not_updated())
     else:
         # Re-query the user to ensure the latest object is returned
-        user = User.query.filter(User.user_name == user_name).first()
+        user = User.find_by_user_name(user_name)
         return user