CASMCMS-9269: Improve Python type annotations in session templates co…

…ntroller code

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 - CASMCMS-9265: Improve Python type annotations in sessions controller code
+- CASMCMS-9269: Improve Python type annotations in session templates controller code
 
 ### Fixed
 - CASMCMS-8965: Update sessions controller to provide error responses consistently and correctly

src/bos/server/controllers/utils.py

@@ -26,6 +26,7 @@
 from urllib.parse import urlparse, urlunparse
 
 import connexion
+from connexion.lifecycle import ConnexionResponse
 import flask
 
 LOGGER = logging.getLogger(__name__)
@@ -73,3 +74,22 @@ def url_for(endpoint, **values):
     # TODO(CASMCMS-1869): there might be a better way to do this by overriding
     # url_adapter in the context or request, see
     # https://github.com/pallets/flask/blob/a74864ec229141784374f1998324d2cbac837295/flask/helpers.py#L302
+
+
+def _400_bad_request(msg: str) -> ConnexionResponse:
+    """
+    ProblemBadRequest
+    """
+    return connexion.problem(
+        status=400,
+        title="Bad Request",
+        detail=msg)
+
+def _404_resource_not_found(resource_type: str, resource_id: str) -> ConnexionResponse:
+    """
+    ProblemResourceNotFound
+    """
+    return connexion.problem(
+        status=404,
+        title="The resource was not found",
+        detail=f"{resource_type} '{resource_id}' does not exist")

src/bos/server/controllers/v2/sessions.py

@@ -23,6 +23,7 @@
 #
 from collections import defaultdict, Counter
 from datetime import datetime, timedelta
+from functools import partial
 import logging
 import re
 from typing import Literal, Optional
@@ -37,6 +38,7 @@
 from bos.common.utils import exc_type_msg, get_current_time, get_current_timestamp, load_timestamp
 from bos.common.values import Phase, Status
 from bos.server import redis_db_utils as dbutils
+from bos.server.controllers.utils import _400_bad_request, _404_resource_not_found
 from bos.server.controllers.v2.boot_set import BootSetStatus, validate_boot_sets
 from bos.server.controllers.v2.components import get_v2_components_data
 from bos.server.controllers.v2.options import OptionsData
@@ -167,7 +169,7 @@ def patch_v2_session(session_id: str) -> tuple[JsonDict, Literal[200]] | Connexi
     session_key = get_tenant_aware_key(session_id, get_tenant_from_header())
     if session_key not in DB:
         LOGGER.warning("Could not find v2 session %s", session_id)
-        return _404_session_not_found(session_id)
+        return _404_session_not_found(resource_id=session_id)  # pylint: disable=redundant-keyword-arg
 
     component = DB.patch(session_key, patch_data_json)
     return component, 200
@@ -187,7 +189,7 @@ def get_v2_session(
     session_key = get_tenant_aware_key(session_id, get_tenant_from_header())
     if session_key not in DB:
         LOGGER.warning("Could not find v2 session %s", session_id)
-        return _404_session_not_found(session_id)
+        return _404_session_not_found(resource_id=session_id)  # pylint: disable=redundant-keyword-arg
     session = DB.get(session_key)
     return session, 200
 
@@ -223,7 +225,7 @@ def delete_v2_session(
     session_key = get_tenant_aware_key(session_id, get_tenant_from_header())
     if session_key not in DB:
         LOGGER.warning("Could not find v2 session %s", session_id)
-        return _404_session_not_found(session_id)
+        return _404_session_not_found(resource_id=session_id)  # pylint: disable=redundant-keyword-arg
     if session_key in STATUS_DB:
         STATUS_DB.delete(session_key)
     return DB.delete(session_key), 204
@@ -270,7 +272,7 @@ def get_v2_session_status(
     session_key = get_tenant_aware_key(session_id, get_tenant_from_header())
     if session_key not in DB:
         LOGGER.warning("Could not find v2 session %s", session_id)
-        return _404_session_not_found(session_id)
+        return _404_session_not_found(resource_id=session_id)  # pylint: disable=redundant-keyword-arg
     session = DB.get(session_key)
     if session.get(
             "status",
@@ -296,7 +298,7 @@ def save_v2_session_status(
     session_key = get_tenant_aware_key(session_id, get_tenant_from_header())
     if session_key not in DB:
         LOGGER.warning("Could not find v2 session %s", session_id)
-        return _404_session_not_found(session_id)
+        return _404_session_not_found(resource_id=session_id)  # pylint: disable=redundant-keyword-arg
     return STATUS_DB.put(session_key, _get_v2_session_status(session_key)), 200
 
 
@@ -440,24 +442,8 @@ def _age_to_timestamp(age: str) -> datetime:
     return get_current_time() - delta
 
 
-def _400_bad_request(msg: str) -> ConnexionResponse:
-    """
-    ProblemBadRequest
-    """
-    return connexion.problem(
-        status=400,
-        title="Bad Request",
-        detail=msg)
-
+_404_session_not_found = partial(_404_resource_not_found, resource_type="Session")
 
-def _404_session_not_found(session_id: str) -> ConnexionResponse:
-    """
-    ProblemResourceNotFound
-    """
-    return connexion.problem(
-        status=404,
-        title="The resource was not found",
-        detail=f"Session '{session_id}' does not exist")
 
 def _409_session_already_exists(session_id: str) -> ConnexionResponse:
     """

src/bos/server/controllers/v2/sessiontemplates.py

@@ -1,7 +1,7 @@
 #
 # MIT License
 #
-# (C) Copyright 2021-2024 Hewlett Packard Enterprise Development LP
+# (C) Copyright 2021-2025 Hewlett Packard Enterprise Development LP
 #
 # Permission is hereby granted, free of charge, to any person obtaining a
 # copy of this software and associated documentation files (the "Software"),
@@ -21,13 +21,18 @@
 # ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 # OTHER DEALINGS IN THE SOFTWARE.
 #
+from functools import partial
 import logging
-import connexion
+from typing import Literal, Optional
+
+from connexion.lifecycle import ConnexionResponse
 
 from bos.common.tenant_utils import get_tenant_from_header, get_tenant_aware_key, \
                                     reject_invalid_tenant
+from bos.common.types import JsonDict
 from bos.common.utils import exc_type_msg
 from bos.server import redis_db_utils as dbutils
+from bos.server.controllers.utils import _400_bad_request, _404_resource_not_found
 from bos.server.schema import validator
 from bos.server.utils import get_request_json
 from .boot_set import validate_boot_sets, validate_sanitize_boot_sets
@@ -63,7 +68,7 @@
 
 @reject_invalid_tenant
 @dbutils.redis_error_handler
-def put_v2_sessiontemplate(session_template_id):  # noqa: E501
+def put_v2_sessiontemplate(session_template_id: str) -> tuple[JsonDict, Literal[200]] | ConnexionResponse:  # noqa: E501
     """PUT /v2/sessiontemplates
 
     Creates a new session template. # noqa: E501
@@ -75,20 +80,15 @@ def put_v2_sessiontemplate(session_template_id):  # noqa: E501
     except Exception as err:
         LOGGER.error("Error parsing PUT '%s' request data: %s",
                      session_template_id, exc_type_msg(err))
-        return connexion.problem(status=400,
-                                 title="Error parsing the data provided.",
-                                 detail=str(err))
+        return _400_bad_request(f"Error parsing the data provided: {err}")
 
     try:
         validate_sanitize_session_template(session_template_id, template_data)
     except Exception as err:
         LOGGER.error("Error creating session template '%s': %s",
                      session_template_id, exc_type_msg(err))
         LOGGER.debug("Full template: %s", template_data)
-        return connexion.problem(
-            status=400,
-            title="The session template could not be created.",
-            detail=str(err))
+        return _400_bad_request(f"The session template could not be created: {err}")
 
     tenant = get_tenant_from_header()
     template_data['tenant'] = tenant
@@ -97,7 +97,7 @@ def put_v2_sessiontemplate(session_template_id):  # noqa: E501
 
 
 @dbutils.redis_error_handler
-def get_v2_sessiontemplates():  # noqa: E501
+def get_v2_sessiontemplates() -> tuple[list[JsonDict], Literal[200]]:  # noqa: E501
     """
     GET /v2/sessiontemplates
 
@@ -111,7 +111,7 @@ def get_v2_sessiontemplates():  # noqa: E501
 
 
 @dbutils.redis_error_handler
-def get_v2_sessiontemplate(session_template_id):
+def get_v2_sessiontemplate(session_template_id: str) -> tuple[JsonDict, Literal[200]] | ConnexionResponse:
     """
     GET /v2/sessiontemplates
 
@@ -123,16 +123,13 @@ def get_v2_sessiontemplate(session_template_id):
                                         get_tenant_from_header())
     if template_key not in DB:
         LOGGER.warning("Session template not found: %s", session_template_id)
-        return connexion.problem(
-            status=404,
-            title="Sessiontemplate could not found.",
-            detail=f"Sessiontemplate {session_template_id} could not be found")
+        return _404_template_not_found(resource_id=session_template_id)  # pylint: disable=redundant-keyword-arg
     template = DB.get(template_key)
     return template, 200
 
 
 @dbutils.redis_error_handler
-def get_v2_sessiontemplatetemplate():
+def get_v2_sessiontemplatetemplate() -> tuple[JsonDict, Literal[200]]:
     """
     GET /v2/sessiontemplatetemplate
 
@@ -145,7 +142,7 @@ def get_v2_sessiontemplatetemplate():
 
 
 @dbutils.redis_error_handler
-def delete_v2_sessiontemplate(session_template_id):
+def delete_v2_sessiontemplate(session_template_id: str) -> tuple[None, Literal[204]] | ConnexionResponse:
     """
     DELETE /v2/sessiontemplates
 
@@ -158,15 +155,12 @@ def delete_v2_sessiontemplate(session_template_id):
                                         get_tenant_from_header())
     if template_key not in DB:
         LOGGER.warning("Session template not found: %s", session_template_id)
-        return connexion.problem(
-            status=404,
-            title="Sessiontemplate could not found.",
-            detail=f"Sessiontemplate {session_template_id} could not be found")
+        return _404_template_not_found(resource_id=session_template_id)  # pylint: disable=redundant-keyword-arg
     return DB.delete(template_key), 204
 
 
 @dbutils.redis_error_handler
-def patch_v2_sessiontemplate(session_template_id):
+def patch_v2_sessiontemplate(session_template_id: str) -> tuple[JsonDict, Literal[200]] | ConnexionResponse:
     """
     PATCH /v2/sessiontemplates
 
@@ -179,35 +173,27 @@ def patch_v2_sessiontemplate(session_template_id):
                                         get_tenant_from_header())
     if template_key not in DB:
         LOGGER.warning("Session template not found: %s", session_template_id)
-        return connexion.problem(
-            status=404,
-            title="Sessiontemplate could not found.",
-            detail=f"Sessiontemplate {session_template_id} could not be found")
+        return _404_template_not_found(resource_id=session_template_id)  # pylint: disable=redundant-keyword-arg
 
     try:
         template_data = get_request_json()
     except Exception as err:
         LOGGER.error("Error parsing PATCH '%s' request data: %s",
                      session_template_id, exc_type_msg(err))
-        return connexion.problem(status=400,
-                                 title="Error parsing the data provided.",
-                                 detail=str(err))
+        return _400_bad_request(f"Error parsing the data provided: {err}")
 
     try:
         validate_sanitize_session_template(session_template_id, template_data)
     except Exception as err:
         LOGGER.error("Error patching session template '%s': %s",
                      session_template_id, exc_type_msg(err))
-        return connexion.problem(
-            status=400,
-            title="The session template could not be patched.",
-            detail=str(err))
+        return _400_bad_request(f"The session template could not be patched: {err}")
 
     return DB.patch(template_key, template_data), 200
 
 
 @dbutils.redis_error_handler
-def validate_v2_sessiontemplate(session_template_id: str):
+def validate_v2_sessiontemplate(session_template_id: str) -> tuple[str, Literal[200]] | ConnexionResponse:
     """
     Validate a V2 session template. Look for missing elements or errors that would prevent
     a session from being launched using this template.
@@ -230,20 +216,20 @@ def validate_v2_sessiontemplate(session_template_id: str):
     return msg, 200
 
 
-def _get_filtered_templates(tenant):
+def _get_filtered_templates(tenant: Optional[str]) -> list[JsonDict]:
     response = DB.get_all()
     if any([tenant]):
         response = [r for r in response if _matches_filter(r, tenant)]
     return response
 
 
-def _matches_filter(data, tenant):
+def _matches_filter(data: JsonDict, tenant: str) -> bool:
     if tenant and tenant != data.get("tenant"):
         return False
     return True
 
 
-def validate_sanitize_session_template(session_template_id, template_data):
+def validate_sanitize_session_template(session_template_id: str, template_data: JsonDict) -> None:
     """
     Used when creating or patching session templates
     """
@@ -259,3 +245,6 @@ def validate_sanitize_session_template(session_template_id, template_data):
     # validate_sanitize_boot_sets()
     for bs in template_data["boot_sets"].values():
         del bs["name"]
+
+
+_404_template_not_found = partial(_404_resource_not_found, resource_type="Session template")