refactor: migrate to flask-smorest with OpenAPI auto-generation

app.py

@@ -1,16 +1,30 @@
 # app.py
-from flask import Flask, request, jsonify
-import jwt
-import requests
 from dotenv import load_dotenv
+from flask import Flask
+from flask import Flask, request, jsonify
+from flask_smorest import Api
 from os import getenv
-from routes.work_item_details import bp as main_bp
+from routes.work_item_details import blp as main_blp
+from routes.whoami import blp as whoami_blp
 from utils.auth import require_auth
-from flask import Flask
+import jwt
+import requests
 
 load_dotenv()
 
 app = Flask(__name__)
+app.config["API_TITLE"] = "JGI Quality API"
+app.config["API_VERSION"] = "1.0"
+app.config["OPENAPI_VERSION"] = "3.0.3"
+app.config["OPENAPI_URL_PREFIX"] = "/"
+app.config["OPENAPI_SWAGGER_UI_PATH"] = "/docs"
+app.config["OPENAPI_SWAGGER_UI_URL"] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"
+
+api = Api(app)
+
+# Register all your blueprints with this, not `app`
+api.register_blueprint(main_blp)
+api.register_blueprint(whoami_blp)
 
 TENANT_ID = getenv("AZURE_TENANT_ID")
 AUDIENCE = getenv("AZURE_CLIENT_ID")

requirements.txt

@@ -1,6 +1,8 @@
 cryptography
 flask
+flask-smorest
 git+https://github.com/Johnson-Gage-Inspection-Inc/qualer-sdk-python.git@f50f852c5af20f4cc46a6dcf8c0ab459060e7504#egg=qualer_sdk
+marshmallow
 msal
 pyjwt
 python-dotenv

routes/whoami.py

@@ -1,13 +1,18 @@
-# routes/whoami.py
-from flask import Blueprint, jsonify, request
+from flask_smorest import Blueprint
+from flask.views import MethodView
 from utils.auth import require_auth
+from flask import request
+from schemas import WhoamiResponse
 
-bp = Blueprint("whoami", __name__)
+blp = Blueprint("whoami", __name__, url_prefix="/")
 
-@bp.route("/whoami", methods=["GET"])
-@require_auth
-def whoami():
-    return jsonify({
-        "user": request.claims.get("preferred_username"),
-        "sub": request.claims.get("sub")
-    })
+
+@blp.route("/whoami")
+class Whoami(MethodView):
+    @require_auth
+    @blp.response(200, WhoamiResponse)
+    def get(self):
+        return {
+            "user": request.claims.get("preferred_username"),
+            "sub": request.claims.get("sub")
+        }

routes/work_item_details.py

@@ -1,7 +1,12 @@
 # routes/work_item_details.py
 from concurrent.futures import ThreadPoolExecutor
-from flask import Blueprint, request, jsonify
+from flask.views import MethodView
+from flask_smorest import Blueprint, abort
+from flask import request
+from schemas import WorkItemDetailsSchema
 from utils.auth import require_auth
+import re
+from os import getenv
 from qualer_sdk import (
     ServiceOrdersApi,
     ServiceOrderItemsApi,
@@ -10,105 +15,75 @@
     ClientAssetsApi,
     ClientAssetAttributesApi)
 
-bp = Blueprint("main", __name__)
+blp = Blueprint("work_item_details", __name__, url_prefix="/")
 
 
-@bp.route("/work_item_details", methods=["GET"])
-@require_auth
-def work_item_details_route():
-    """Fetches work item details for a given work item number."""
-
-    def get_work_item_details_for_tus(work_item_number):
-        """
-        Fetches work item details for a given work item number.
-
-        Args:
-            work_item_number (str): The work item number to fetch details for.
-
-        Returns:
-            list: A dictionary containing work item details.
-        """
-        from os import getenv
-        import re
+def get_work_item_details_for_tus(work_item_number):
+    pattern = r"^(56561-)?\d{6}(\.\d{2})?(-\d{2})(R\d{1,2})?$"
+    if not re.match(pattern, work_item_number):
+        raise ValueError("Invalid work item number format.")
 
-        pattern = r"^(56561-)?\d{6}(\.\d{2})?(-\d{2})(R\d{1,2})?$"
-        if not re.match(pattern, work_item_number):
-            raise ValueError("Invalid work item number format.")
+    config = Configuration()
+    config.host = "https://jgiquality.qualer.com"
 
-        config = Configuration()
-        config.host = "https://jgiquality.qualer.com"
+    client = ApiClient(configuration=config)
+    client.default_headers["Authorization"] = f'Api-Token {getenv("QUALER_API_KEY")}'
 
-        client = ApiClient(configuration=config)
-        bearer_token = f'Api-Token {getenv("QUALER_API_KEY")}'
-        client.default_headers["Authorization"] = bearer_token
+    soi_api = ServiceOrderItemsApi(client)
+    work_items = soi_api.get_work_items_0(work_item_number=work_item_number)
 
-        soi_api = ServiceOrderItemsApi(client)
-        work_items = soi_api.get_work_items_0(
-            work_item_number=work_item_number
-        )
-        if len(work_items) == 0:
-            raise ValueError(
-                "No work items found for the given work item number."
-                )
-        if len(work_items) > 1:
-            raise ValueError(
-                "Multiple work items found for the given work item number."
-                )
-        # Assuming we only need the first work item
-        item = work_items[0]
+    if len(work_items) == 0:
+        raise ValueError("No work items found for the given work item number.")
+    if len(work_items) > 1:
+        raise ValueError("Multiple work items found for the given work item number.")
 
-        service_order_id = item.service_order_id
-        asset_id = item.asset_id
-        client_company_id = item.client_company_id
+    item = work_items[0]
+    service_order_id = item.service_order_id
+    asset_id = item.asset_id
+    client_company_id = item.client_company_id
 
-        for field in [service_order_id, asset_id, client_company_id]:
-            if field is None:
-                raise ValueError(f"Missing required field: {field}")
+    for field in [service_order_id, asset_id, client_company_id]:
+        if field is None:
+            raise ValueError(f"Missing required field: {field}")
 
-        # Fetch the client asset, attributes, and service order in parallel
-        with ThreadPoolExecutor() as executor:
-            future_client_asset = executor.submit(
-                ClientAssetsApi(client).get_asset,
-                asset_id=asset_id
-            )
-            future_attributes = executor.submit(
-                ClientAssetAttributesApi(client).get_asset_attributes,
-                asset_id=asset_id
-            )
-            future_service_order = executor.submit(
-                ServiceOrdersApi(client).get_work_order,
-                service_order_id=service_order_id
-            )
+    with ThreadPoolExecutor() as executor:
+        future_client_asset = executor.submit(ClientAssetsApi(client).get_asset, asset_id=asset_id)
+        future_attributes = executor.submit(ClientAssetAttributesApi(client).get_asset_attributes, asset_id=asset_id)
+        future_service_order = executor.submit(ServiceOrdersApi(client).get_work_order, service_order_id=service_order_id)
 
-            client_asset = future_client_asset.result()
-            client_asset_attributes = future_attributes.result()
-            service_order = future_service_order.result()
+        client_asset = future_client_asset.result()
+        client_asset_attributes = future_attributes.result()
+        service_order = future_service_order.result()
 
-        # Assemble the response
-        return {
-            "clientCompanyId": client_company_id,
-            "serviceOrderId": service_order_id,
-            "assetId": asset_id,
-            "certificateNumber": item.certificate_number,
-            "assetName": client_asset.asset_name,
-            "assetMaker": client_asset.asset_maker,
-            "assetTag": client_asset.asset_tag,
-            "serialNumber": client_asset.serial_number,
-            "manufacturerPartNumber": client_asset.manufacturer_part_number,
-            "categoryName": client_asset.category_name,
-            "rootCategoryName": client_asset.root_category_name,
-            "productManufacturer": client_asset.product_manufacturer,
-            "productName": client_asset.product_name,
-            "purchaseOrderNumber": service_order.po_number,
-            "assetAttributes": client_asset_attributes
-        }
+    return {
+        "clientCompanyId": client_company_id,
+        "serviceOrderId": service_order_id,
+        "assetId": asset_id,
+        "certificateNumber": item.certificate_number,
+        "assetName": client_asset.asset_name,
+        "assetMaker": client_asset.asset_maker,
+        "assetTag": client_asset.asset_tag,
+        "serialNumber": client_asset.serial_number,
+        "manufacturerPartNumber": client_asset.manufacturer_part_number,
+        "categoryName": client_asset.category_name,
+        "rootCategoryName": client_asset.root_category_name,
+        "productManufacturer": client_asset.product_manufacturer,
+        "productName": client_asset.product_name,
+        "purchaseOrderNumber": service_order.po_number,
+        "assetAttributes": client_asset_attributes
+    }
 
-    work_item_number = request.args.get("workItemNumber")
-    if not work_item_number:
-        return jsonify({"error": "Missing workItemNumber"}), 400
 
-    try:
-        details = get_work_item_details_for_tus(work_item_number)
-        return jsonify(details)
-    except Exception as e:
-        return jsonify({"error": str(e)}), 500
+@blp.route("/work_item_details")
+class WorkItemDetails(MethodView):
+    @require_auth
+    @blp.response(200, WorkItemDetailsSchema)
+    def get(self):
+        work_item_number = request.args.get("workItemNumber")
+        if not work_item_number:
+            abort(400, message="Missing workItemNumber")
+
+        try:
+            return get_work_item_details_for_tus(work_item_number)
+        except Exception as e:
+            abort(500, message=str(e))

schemas.py

@@ -0,0 +1,22 @@
+from marshmallow import Schema, fields
+
+class WorkItemDetailsSchema(Schema):
+    clientCompanyId = fields.Int(required=True)
+    serviceOrderId = fields.Int(required=True)
+    assetId = fields.Int(required=True)
+    certificateNumber = fields.Str()
+    assetName = fields.Str()
+    assetMaker = fields.Str()
+    assetTag = fields.Str()
+    serialNumber = fields.Str()
+    manufacturerPartNumber = fields.Str()
+    categoryName = fields.Str()
+    rootCategoryName = fields.Str()
+    productManufacturer = fields.Str()
+    productName = fields.Str()
+    purchaseOrderNumber = fields.Str()
+    assetAttributes = fields.Dict()
+
+class WhoamiResponse(Schema):
+    user = fields.String(required=True)
+    sub = fields.String(required=True)

tests/conftest.py

@@ -1,7 +1,6 @@
-from app import app as flask_app
 from flask import Flask
-from routes.work_item_details import bp as work_item_details_bp
-from routes.whoami import bp as whoami_bp
+from routes.work_item_details import blp as work_item_details_blp
+from routes.whoami import blp as whoami_blp
 import pytest
 from get_token import get_access_token
 
@@ -16,8 +15,8 @@ def client():
     app = Flask(__name__)
     app.config["TESTING"] = True
 
-    app.register_blueprint(work_item_details_bp)
-    app.register_blueprint(whoami_bp)
+    app.register_blueprint(work_item_details_blp)
+    app.register_blueprint(whoami_blp)
 
     with app.test_client() as client:
         yield client