Fixing compilation errors on pytest and some lint fixes

forms-flow-api/src/formsflow_api/app.py

@@ -1,3 +1,7 @@
+"""App inititlization.
+
+Initialize app and the dependencies.
+"""
 import logging
 import os
 from flask import Flask, request
@@ -33,15 +37,17 @@ def create_app(run_mode=os.getenv("FLASK_ENV", "production")):
     )
     app.logger = flask_logger
     app.logger = logging.getLogger("app")
-    ch = logging.StreamHandler()
 
-    ch.setFormatter(CustomFormatter())
-    app.logger.handlers = [ch]
+    stream_handler = logging.StreamHandler()
+    stream_handler.setFormatter(CustomFormatter())
+    app.logger.handlers = [stream_handler]
     app.logger.propagate = False
     logging.log.propagate = False
-    with open("logo.txt") as f:
+
+    with open("logo.txt", encoding="utf-8") as f:
         contents = f.read()
         print(contents)
+
     app.logger.info("Welcome to formsflow-API server...!")
     db.init_app(app)
     ma.init_app(app)

forms-flow-api/src/formsflow_api/config.py

@@ -123,7 +123,6 @@ class TestConfig(_Config):  # pylint: disable=too-few-public-methods
     JWT_OIDC_TEST_JWKS_CACHE_TIMEOUT = 6000
 
 
-
 class ProdConfig(_Config):  # pylint: disable=too-few-public-methods
     """Production environment configuration."""
 

forms-flow-api/src/formsflow_api/models/application_history.py

@@ -8,6 +8,7 @@
 
 class ApplicationHistory(ApplicationAuditDateTimeMixin, BaseModel, db.Model):
     """This class manages application audit against each form."""
+
     __tablename__ = "application_audit"
     id = db.Column(db.Integer, primary_key=True)
     application_id = db.Column(db.Integer, nullable=False)

forms-flow-api/src/formsflow_api/resources/__init__.py

@@ -48,7 +48,8 @@ def handle_auth_error(error: AuthError):
     return (
         {
             "type": "Invalid Token Error",
-            "message": "Access to formsflow.ai API Denied. Check if the bearer token is passed for Authorization or has expired.",
+            "message": "Access to formsflow.ai API Denied. Check if the bearer token is passed for '\
+                'Authorization or has expired.",
         },
         error.status_code,
         {"Access-Control-Allow-Origin": ALLOW_ALL_ORIGINS},

forms-flow-api/src/formsflow_api/resources/application.py

@@ -27,7 +27,7 @@ class ApplicationsResource(Resource):
     @staticmethod
     @auth.require
     @profiletime
-    def get():
+    def get(): # pylint:disable=too-many-locals
         """Get applications
         Fetch applications list based on parameters
         :params Id: list the application for particular id
@@ -40,7 +40,7 @@ def get():
         :params limit: to retrieve limit for each page
         :params orderBy: Name of column to order by (default: created)
         """
-        try: 
+        try:
             dict_data = ApplicationListRequestSchema().load(request.args) or {}
             page_no = dict_data.get("page_no")
             limit = dict_data.get("limit")
@@ -93,8 +93,8 @@ def get():
                     application_status=application_status,
                 )
             application_schema = ApplicationService.apply_custom_attributes(
-                    application_schema=application_schema_dump
-                )
+                application_schema=application_schema_dump
+            )
             return (
                 (
                     {
@@ -143,12 +143,11 @@ def get(application_id):
                     ApplicationService.apply_custom_attributes(application_schema_dump),
                     status,
                 )
-            else:
-                application, status = ApplicationService.get_application_by_user(
-                    application_id=application_id,
-                    user_id=g.token_info.get("preferred_username"),
-                )
-                return (ApplicationService.apply_custom_attributes(application), status)
+            application, status = ApplicationService.get_application_by_user(
+                application_id=application_id,
+                user_id=g.token_info.get("preferred_username"),
+            )
+            return (ApplicationService.apply_custom_attributes(application), status)
         except BusinessException as err:
             return err.error, err.status_code
 
@@ -167,7 +166,7 @@ def put(application_id):
                 application_id=application_id, data=dict_data
             )
             return "Updated successfully", HTTPStatus.OK
-        except BaseException as submission_err:
+        except BaseException as submission_err: # pylint: disable=broad-except
             response, status = {
                 "type": "Bad request error",
                 "message": "Invalid request data",
@@ -231,18 +230,18 @@ def get(form_id):
                 ),
                 HTTPStatus.OK,
             )
-        else:
-            return (
-                (
-                    {
-                        "applications": application_schema,
-                        "totalCount": application_count,
-                        "limit": limit,
-                        "pageNo": page_no,
-                    }
-                ),
-                HTTPStatus.OK,
-            )
+
+        return (
+            (
+                {
+                    "applications": application_schema,
+                    "totalCount": application_count,
+                    "limit": limit,
+                    "pageNo": page_no,
+                }
+            ),
+            HTTPStatus.OK,
+        )
 
 
 @cors_preflight("POST,OPTIONS")
@@ -267,7 +266,7 @@ def post():
             )
             response, status = application_schema.dump(application), HTTPStatus.CREATED
             return response, status
-        except BaseException as application_err:
+        except BaseException as application_err: # pylint:disable=broad-except
             response, status = {
                 "type": "Bad request error",
                 "message": "Invalid application request passed",
@@ -286,7 +285,7 @@ class ProcessMapperResourceByApplicationId(Resource):
     @auth.require
     @profiletime
     def get(application_id):
-
+        """Get process by application id."""
         try:
             return (
                 ApplicationService.get_application_form_mapper_by_id(application_id),
@@ -305,7 +304,7 @@ class ApplicationResourceByApplicationStatus(Resource):
     @auth.require
     @profiletime
     def get():
-
+        """Get status list."""
         try:
             return (
                 ApplicationService.get_all_application_status(),

forms-flow-api/src/formsflow_api/resources/application_history.py

@@ -67,7 +67,7 @@ def post(application_id):
             current_app.logger.error(err)
             return response, status
 
-        except BaseException as application_err:
+        except BaseException as application_err:  # pylint: disable=broad-except
             response, status = {
                 "type": "Invalid Request Object",
                 "message": "Invalid Request Object Passed ",
@@ -76,5 +76,5 @@ def post(application_id):
 
             current_app.logger.warning(response)
             current_app.logger.warning(application_err)
-        finally:
-            return response, status
+
+        return response, status

forms-flow-api/src/formsflow_api/resources/dashboards.py

@@ -26,7 +26,7 @@ class DashboardList(Resource):
     @API.doc("list_dashboards")
     @auth.require
     @profiletime
-    def get(self):
+    def get(self):  # pylint: disable=no-self-use
         """List all dashboards"""
         if request.args:
             dict_data = ApplicationListReqSchema().load(request.args)
@@ -38,13 +38,14 @@ def get(self):
         response = analytics_service.get_request(
             url_path="dashboards", page_no=page_no, limit=limit
         )
+
         if response == "unauthorized":
             return {"message": "Dashboards not available"}, HTTPStatus.NOT_FOUND
-        elif response is None:
+        if response is None:
             return {"message": "Error"}, HTTPStatus.SERVICE_UNAVAILABLE
-        else:
-            assert response != None
-            return response, HTTPStatus.OK
+
+        assert response is not None
+        return response, HTTPStatus.OK
 
 
 @cors_preflight("GET,OPTIONS")
@@ -55,7 +56,7 @@ class DashboardDetail(Resource):
     @API.doc("get_dashboard")
     @auth.require
     @profiletime
-    def get(self, dashboard_id):
+    def get(self, dashboard_id):  # pylint: disable=no-self-use
         """Get a dashboard with given dashboard_id"""
         try:
             available_dashboards = re.findall(
@@ -73,8 +74,8 @@ def get(self, dashboard_id):
             )
             if response == "unauthorized":
                 return {"message": "Dashboard not found"}, HTTPStatus.NOT_FOUND
-            elif response is None:
+            if response is None:
                 return {"message": "Error"}, HTTPStatus.SERVICE_UNAVAILABLE
-            else:
-                assert response != None
-                return response, HTTPStatus.OK
+
+            assert response is not None
+            return response, HTTPStatus.OK

forms-flow-api/src/formsflow_api/resources/groups.py

@@ -53,36 +53,39 @@ def get():
                     return {
                         "message": "No Dashboard authorized Group found"
                     }, HTTPStatus.NOT_FOUND
-                else:
-                    for group in dashboard_group_list:
-                        group["dashboards"] = (
-                            client.get_request(url_path=f"groups/{group['id']}")
-                            .get("attributes")
-                            .get("dashboards")
-                        )
-                    return dashboard_group_list, HTTPStatus.OK
+
+                for group in dashboard_group_list:  # pylint:disable=redefined-outer-name
+                    group["dashboards"] = (
+                        client.get_request(url_path=f"groups/{group['id']}")
+                        .get("attributes")
+                        .get("dashboards")
+                    )
+                return dashboard_group_list, HTTPStatus.OK
+        return None, HTTPStatus.NOT_FOUND
 
 
 @cors_preflight("GET,PUT,OPTIONS")
-@API.route("/<string:id>", methods=["GET", "PUT", "OPTIONS"])
+@API.route("/<string:group_id>", methods=["GET", "PUT", "OPTIONS"])
 class KeycloakDashboardGroupDetail(Resource):
+    """Keycloak dashboard group detail class."""
+
     @staticmethod
     @auth.require
     @profiletime
-    def get(id):
+    def get(group_id):
         """GET request to fetch groups details API
         :params str id: group-id of Keycloak Dashboard Authorized groups
         """
         client = KeycloakAdminAPIService()
-        response = client.get_request(url_path=f"groups/{id}")
+        response = client.get_request(url_path=f"groups/{group_id}")
         if response is None:
-            return {"message": f"Group - {id} not found"}, HTTPStatus.NOT_FOUND
+            return {"message": f"Group - {group_id} not found"}, HTTPStatus.NOT_FOUND
         return response
 
     @staticmethod
     @auth.require
     @profiletime
-    def put(id):
+    def put(group_id):
         """Update request to update dashboard details
         :params str id: group-id of Keycloak Dashboard Authorized groups
         """
@@ -91,22 +94,21 @@ def put(id):
         try:
             dict_data = KeycloakDashboardGroupSchema().load(group_json)
 
-            dashboard_id_details = client.get_request(url_path=f"groups/{id}")
-            if dashboard_id_details == None:
-                return {"message": f"Group - {id} not found"}, HTTPStatus.NOT_FOUND
-            else:
-                dashboard_id_details["attributes"]["dashboards"] = [
-                    str(dict_data["dashboards"])
-                ]
-                response = client.update_request(
-                    url_path=f"groups/{id}", data=dashboard_id_details
-                )
-                if response is None:
-                    return {
-                        "message": f"Group - {id} not updated successfully"
-                    }, HTTPStatus.SERVICE_UNAVAILABLE
-                else:
-                    return response
+            dashboard_id_details = client.get_request(url_path=f"groups/{group_id}")
+            if dashboard_id_details is None:
+                return {"message": f"Group - {group_id} not found"}, HTTPStatus.NOT_FOUND
+
+            dashboard_id_details["attributes"]["dashboards"] = [
+                str(dict_data["dashboards"])
+            ]
+            response = client.update_request(
+                url_path=f"groups/{group_id}", data=dashboard_id_details
+            )
+            if response is None:
+                return {
+                    "message": f"Group - {group_id} not updated successfully"
+                }, HTTPStatus.SERVICE_UNAVAILABLE
+            return response
         except ValidationError as err:
             pprint(err.messages)
             return {"message": "Invalid Request Object format"}, HTTPStatus.BAD_REQUEST