Use httpstatus module

backend/ibutsu_server/__init__.py

@@ -1,4 +1,5 @@
 import os
+from http import HTTPStatus
 from importlib import import_module
 from pathlib import Path
 from typing import Any, Optional
@@ -130,38 +131,38 @@ def get_app(**extra_config):
 
     @app.route("/")
     def index():
-        return redirect("/api/ui/", code=302)
+        return redirect("/api/ui/", code=HTTPStatus.FOUND)
 
     @app.route("/admin/run-task", methods=["POST"])
     def run_task():
         # get params
         params = request.get_json(force=True, silent=True)
         if not params:
-            return "Bad request", 400
+            return HTTPStatus.BAD_REQUEST.phrase, HTTPStatus.BAD_REQUEST
         # get user info
         token = params.get("token")
         if not token:
-            return "Unauthorized", 401
+            return HTTPStatus.UNAUTHORIZED.phrase, HTTPStatus.UNAUTHORIZED
         user_id = decode_token(token).get("sub")
         if not user_id:
-            return "Unauthorized", 401
+            return HTTPStatus.UNAUTHORIZED.phrase, HTTPStatus.UNAUTHORIZED
         user = User.query.get(user_id)
         if not user or not user.is_superadmin:
-            return "Forbidden", 403
+            return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
         # get task info
         task_path = params.get("task")
         task_params = params.get("params", {})
         if not task_path:
-            return "Bad request", 400
+            return HTTPStatus.BAD_REQUEST.phrase, HTTPStatus.BAD_REQUEST
         task_module, task_name = task_path.split(".", 2)
         try:
             mod = import_module(f"ibutsu_server.tasks.{task_module}")
         except ImportError:
-            return "Not found", 404
+            return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND
         if not hasattr(mod, task_name):
-            return "Not found", 404
+            return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND
         task = getattr(mod, task_name)
         task.delay(**task_params)
-        return "Accepted", 202
+        return HTTPStatus.ACCEPTED.phrase, HTTPStatus.ACCEPTED
 
     return app.app

backend/ibutsu_server/constants.py

@@ -1,3 +1,5 @@
+from http import HTTPStatus
+
 LOCALHOST = "127.0.0.1"
 
 OAUTH_CONFIG = {
@@ -394,3 +396,5 @@
         "type": "widget",
     },
 }
+
+RESPONSE_JSON_REQ = f"{HTTPStatus.BAD_REQUEST.phrase} JSON required", HTTPStatus.BAD_REQUEST

backend/ibutsu_server/controllers/admin/project_controller.py

@@ -1,6 +1,9 @@
+from http import HTTPStatus
+
 import connexion
 from flask import abort
 
+from ibutsu_server.constants import RESPONSE_JSON_REQ
 from ibutsu_server.db.base import session
 from ibutsu_server.db.models import Group, Project, User
 from ibutsu_server.filters import convert_filter
@@ -19,23 +22,23 @@ def admin_add_project(project=None, token_info=None, user=None):
     """
     check_user_is_admin(user)
     if not connexion.request.is_json:
-        return "Bad request, JSON required", 400
+        return RESPONSE_JSON_REQ
     project = Project.from_dict(**connexion.request.get_json())
     # check if project already exists
     if project.id and Project.query.get(project.id):
-        return f"Project id {project.id} already exist", 400
+        return f"Project id {project.id} already exist", HTTPStatus.BAD_REQUEST
     user = User.query.get(user)
     if project.group_id:
         # check if the group exists
         group = Group.query.get(project.group_id)
         if not group:
-            return f"Group id {project.group_id} doesn't exist", 400
+            return f"Group id {project.group_id} doesn't exist", HTTPStatus.BAD_REQUEST
     if user:
         project.owner = user
         project.users.append(user)
     session.add(project)
     session.commit()
-    return project.to_dict(), 201
+    return project.to_dict(), HTTPStatus.CREATED
 
 
 @validate_uuid
@@ -52,7 +55,7 @@ def admin_get_project(id_, token_info=None, user=None):
     if not project:
         project = Project.query.filter(Project.name == id_).first()
     if not project:
-        abort(404)
+        abort(HTTPStatus.NOT_FOUND)
     return project.to_dict(with_owner=True)
 
 
@@ -95,7 +98,7 @@ def admin_get_project_list(
     total_items = query.count()
     total_pages = (total_items // page_size) + (1 if total_items % page_size > 0 else 0)
     if offset > 9223372036854775807:  # max value of bigint
-        return "The page number is too big.", 400
+        return "The page number is too big.", HTTPStatus.BAD_REQUEST
     projects = query.offset(offset).limit(page_size).all()
     return {
         "projects": [project.to_dict(with_owner=True) for project in projects],
@@ -121,13 +124,13 @@ def admin_update_project(id_, project=None, body=None, token_info=None, user=Non
     """
     check_user_is_admin(user)
     if not connexion.request.is_json:
-        return "Bad request, JSON required", 400
+        return RESPONSE_JSON_REQ
     if not is_uuid(id_):
         id_ = convert_objectid_to_uuid(id_)
     project = Project.query.get(id_)
 
     if not project:
-        abort(404)
+        abort(HTTPStatus.NOT_FOUND)
 
     # Grab the fields from the request
     project_dict = connexion.request.get_json()
@@ -159,10 +162,10 @@ def admin_delete_project(id_, token_info=None, user=None):
     """Delete a single project"""
     check_user_is_admin(user)
     if not is_uuid(id_):
-        return f"Project ID {id_} is not in UUID format", 400
+        return f"Project ID {id_} is not in UUID format", HTTPStatus.BAD_REQUEST
     project = Project.query.get(id_)
     if not project:
-        abort(404)
+        abort(HTTPStatus.NOT_FOUND)
     session.delete(project)
     session.commit()
-    return "OK", 200
+    return HTTPStatus.OK.phrase, HTTPStatus.OK

backend/ibutsu_server/controllers/admin/user_controller.py

@@ -1,6 +1,9 @@
+from http import HTTPStatus
+
 import connexion
 from flask import abort
 
+from ibutsu_server.constants import RESPONSE_JSON_REQ
 from ibutsu_server.db.base import session
 from ibutsu_server.db.models import Project, User
 from ibutsu_server.filters import convert_filter
@@ -27,7 +30,7 @@ def admin_get_user(id_, token_info=None, user=None):
     check_user_is_admin(user)
     requested_user = User.query.get(id_)
     if not requested_user:
-        abort(404)
+        abort(HTTPStatus.NOT_FOUND)
     return _hide_sensitive_fields(requested_user.to_dict(with_projects=True))
 
 
@@ -63,27 +66,27 @@ def admin_add_user(new_user=None, token_info=None, user=None):
     """Create a new user in the system"""
     check_user_is_admin(user)
     if not connexion.request.is_json:
-        return "Bad request, JSON required", 400
+        return RESPONSE_JSON_REQ
     new_user = User.from_dict(**connexion.request.get_json())
     user_exists = User.query.filter_by(email=new_user.email).first()
     if user_exists:
-        return f"The user with email {new_user.email} already exists", 400
+        return f"The user with email {new_user.email} already exists", HTTPStatus.BAD_REQUEST
     session.add(new_user)
     session.commit()
-    return _hide_sensitive_fields(new_user.to_dict()), 201
+    return _hide_sensitive_fields(new_user.to_dict()), HTTPStatus.CREATED
 
 
 @validate_uuid
 def admin_update_user(id_, body=None, user_info=None, token_info=None, user=None):
     """Update a single user in the system"""
     check_user_is_admin(user)
     if not connexion.request.is_json:
-        return "Bad request, JSON required", 400
+        return RESPONSE_JSON_REQ
     user_dict = connexion.request.get_json()
     projects = user_dict.pop("projects", [])
     requested_user = User.query.get(id_)
     if not requested_user:
-        abort(404)
+        abort(HTTPStatus.NOT_FOUND)
     requested_user.update(user_dict)
     requested_user.projects = [Project.query.get(project["id"]) for project in projects]
     session.add(requested_user)
@@ -97,7 +100,7 @@ def admin_delete_user(id_, token_info=None, user=None):
     check_user_is_admin(user)
     requested_user = User.query.get(id_)
     if not requested_user:
-        abort(404)
+        abort(HTTPStatus.NOT_FOUND)
     session.delete(requested_user)
     session.commit()
-    return "OK", 200
+    return HTTPStatus.OK.phrase, HTTPStatus.OK

backend/ibutsu_server/controllers/artifact_controller.py

@@ -1,5 +1,6 @@
 import json
 from datetime import datetime
+from http import HTTPStatus
 
 import connexion
 import magic
@@ -16,9 +17,9 @@ def _build_artifact_response(id_):
     """Build a response for the artifact"""
     artifact = Artifact.query.get(id_)
     if not artifact:
-        return "Not Found", 404
+        return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND
     # Create a response with the contents of this file
-    response = make_response(artifact.content, 200)
+    response = make_response(artifact.content, HTTPStatus.OK)
     # Set the content type and the file name
     file_type = magic.from_buffer(artifact.content, mime=True)
     response.headers["Content-Type"] = file_type
@@ -36,9 +37,9 @@ def view_artifact(id_, token_info=None, user=None):
     """
     artifact, response = _build_artifact_response(id_)
     if artifact.result and not project_has_user(artifact.result.project, user):
-        return "Forbidden", 403
+        return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
     elif artifact.run and not project_has_user(artifact.run.project, user):
-        return "Forbidden", 403
+        return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
     return response
 
 
@@ -53,7 +54,7 @@ def download_artifact(id_, token_info=None, user=None):
     """
     artifact, response = _build_artifact_response(id_)
     if not project_has_user(artifact.result.project, user):
-        return "Forbidden", 403
+        return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
     response.headers["Content-Disposition"] = f"attachment; filename={artifact.filename}"
     return response
 
@@ -69,9 +70,9 @@ def get_artifact(id_, token_info=None, user=None):
     """
     artifact = Artifact.query.get(id_)
     if not artifact:
-        return "Not Found", 404
+        return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND
     if not project_has_user(artifact.result.project, user):
-        return "Forbidden", 403
+        return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
     return artifact.to_dict()
 
 
@@ -129,12 +130,12 @@ def upload_artifact(body, token_info=None, user=None):
     result_id = body.get("result_id") or body.get("resultId")
     run_id = body.get("run_id") or body.get("runId")
     if result_id and not is_uuid(result_id):
-        return f"Result ID {result_id} is not in UUID format", 400
+        return f"Result ID {result_id} is not in UUID format", HTTPStatus.BAD_REQUEST
     if run_id and not is_uuid(run_id):
-        return f"Run ID {run_id} is not in UUID format", 400
+        return f"Run ID {run_id} is not in UUID format", HTTPStatus.BAD_REQUEST
     result = Result.query.get(result_id)
     if result and not project_has_user(result.project, user):
-        return "Forbidden", 403
+        return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
     filename = body.get("filename")
     additional_metadata = body.get("additional_metadata", {})
     file_ = connexion.request.files["file"]
@@ -150,9 +151,9 @@ def upload_artifact(body, token_info=None, user=None):
             try:
                 additional_metadata = json.loads(additional_metadata)
             except (ValueError, TypeError):
-                return "Bad request, additionalMetadata is not valid JSON", 400
+                return "Bad request, additionalMetadata is not valid JSON", HTTPStatus.BAD_REQUEST
         if not isinstance(additional_metadata, dict):
-            return "Bad request, additionalMetadata is not a JSON object", 400
+            return "Bad request, additionalMetadata is not a JSON object", HTTPStatus.BAD_REQUEST
         data["additionalMetadata"] = additional_metadata
     # Reset the file pointer
     file_.seek(0)
@@ -175,7 +176,7 @@ def upload_artifact(body, token_info=None, user=None):
 
     session.add(artifact)
     session.commit()
-    return artifact.to_dict(), 201
+    return artifact.to_dict(), HTTPStatus.CREATED
 
 
 @validate_uuid
@@ -189,9 +190,9 @@ def delete_artifact(id_, token_info=None, user=None):
     """
     artifact = Artifact.query.get(id_)
     if not artifact:
-        return "Not Found", 404
+        return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND
     if not project_has_user(artifact.result.project, user):
-        return "Forbidden", 403
+        return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
     session.delete(artifact)
     session.commit()
-    return "OK", 200
+    return HTTPStatus.OK.phrase, HTTPStatus.OK