refactor: reduce cognitive complexity in additional OpenAPI methods

- Refactored _resolve_openapi_config method to reduce complexity from 17 to 15
  - Split into 4 helper methods for better maintainability
  - Improved separation of concerns for config resolution logic

- Refactored get_openapi_schema method in example file to reduce complexity from 23 to 15
  - Split into 6 helper methods with clear responsibilities
  - Enhanced readability and maintainability of schema fixing logic

- All OpenAPI tests continue to pass (220/220)
- Example functionality validated and working correctly

Author: oyiz-michael

aws_lambda_powertools/event_handler/api_gateway.py

@@ -1812,22 +1812,36 @@ def _resolve_openapi_config(self, **kwargs) -> dict[str, Any]:
         # See: https://github.com/aws-powertools/powertools-lambda-python/issues/6122
         resolved = {}
 
-        # Handle title with fallback
+        # Handle fields with specific default value checks
+        self._resolve_title_config(resolved, kwargs)
+        self._resolve_version_config(resolved, kwargs)
+        self._resolve_openapi_version_config(resolved, kwargs)
+
+        # Resolve other fields with simple fallbacks
+        self._resolve_remaining_config_fields(resolved, kwargs)
+
+        return resolved
+
+    def _resolve_title_config(self, resolved: dict[str, Any], kwargs: dict[str, Any]) -> None:
+        """Resolve title configuration with fallback to openapi_config."""
         resolved["title"] = kwargs["title"]
         if kwargs["title"] == DEFAULT_OPENAPI_TITLE and self.openapi_config.title:
             resolved["title"] = self.openapi_config.title
 
-        # Handle version with fallback
+    def _resolve_version_config(self, resolved: dict[str, Any], kwargs: dict[str, Any]) -> None:
+        """Resolve version configuration with fallback to openapi_config."""
         resolved["version"] = kwargs["version"]
         if kwargs["version"] == DEFAULT_API_VERSION and self.openapi_config.version:
             resolved["version"] = self.openapi_config.version
 
-        # Handle openapi_version with fallback
+    def _resolve_openapi_version_config(self, resolved: dict[str, Any], kwargs: dict[str, Any]) -> None:
+        """Resolve openapi_version configuration with fallback to openapi_config."""
         resolved["openapi_version"] = kwargs["openapi_version"]
         if kwargs["openapi_version"] == DEFAULT_OPENAPI_VERSION and self.openapi_config.openapi_version:
             resolved["openapi_version"] = self.openapi_config.openapi_version
 
-        # Resolve other fields with fallbacks
+    def _resolve_remaining_config_fields(self, resolved: dict[str, Any], kwargs: dict[str, Any]) -> None:
+        """Resolve remaining configuration fields with simple fallbacks."""
         resolved.update({
             "summary": kwargs["summary"] or self.openapi_config.summary,
             "description": kwargs["description"] or self.openapi_config.description,
@@ -1842,8 +1856,6 @@ def _resolve_openapi_config(self, **kwargs) -> dict[str, Any]:
             "openapi_extensions": kwargs["openapi_extensions"] or self.openapi_config.openapi_extensions,
         })
 
-        return resolved
-
     def _build_base_openapi_structure(self, config: dict[str, Any]) -> dict[str, Any]:
         """Build the base OpenAPI structure with info, servers, and security."""
         openapi_version = self._determine_openapi_version(config["openapi_version"])

examples/event_handler/openapi_schema_fix_example.py

@@ -49,38 +49,69 @@ def get_openapi_schema(self, **kwargs):
         schema_dict = schema.model_dump(by_alias=True)
 
         # Find all multipart/form-data references that might be missing
-        missing_refs = []
-        paths = schema_dict.get("paths", {})
-        for path_item in paths.values():
-            for _method, operation in path_item.items():
-                if not isinstance(operation, dict):
-                    continue
-
-                if "requestBody" not in operation:
-                    continue
-
-                req_body = operation.get("requestBody", {})
-                content = req_body.get("content", {})
-                multipart = content.get("multipart/form-data", {})
-                schema_ref = multipart.get("schema", {})
-
-                if "$ref" in schema_ref:
-                    ref = schema_ref["$ref"]
-                    if ref.startswith("#/components/schemas/"):
-                        component_name = ref[len("#/components/schemas/") :]
-
-                        # Check if the component exists
-                        components = schema_dict.get("components", {})
-                        schemas = components.get("schemas", {})
-
-                        if component_name not in schemas:
-                            missing_refs.append((component_name, ref))
+        missing_refs = self._find_missing_component_references(schema_dict)
 
         # If no missing references, return the original schema
         if not missing_refs:
             return schema
 
         # Add missing components to the schema
+        self._add_missing_components(schema_dict, missing_refs)
+
+        # Rebuild the schema with the added components
+        return schema.__class__(**schema_dict)
+
+    def _find_missing_component_references(self, schema_dict: dict) -> list[tuple[str, str]]:
+        """Find all missing component references in multipart/form-data schemas."""
+        missing_refs = []
+        paths = schema_dict.get("paths", {})
+        
+        for path_item in paths.values():
+            self._check_path_item_for_missing_refs(path_item, schema_dict, missing_refs)
+        
+        return missing_refs
+
+    def _check_path_item_for_missing_refs(
+        self, 
+        path_item: dict, 
+        schema_dict: dict, 
+        missing_refs: list[tuple[str, str]]
+    ) -> None:
+        """Check a single path item for missing component references."""
+        for _method, operation in path_item.items():
+            if not isinstance(operation, dict) or "requestBody" not in operation:
+                continue
+            
+            self._check_operation_for_missing_refs(operation, schema_dict, missing_refs)
+
+    def _check_operation_for_missing_refs(
+        self, 
+        operation: dict, 
+        schema_dict: dict, 
+        missing_refs: list[tuple[str, str]]
+    ) -> None:
+        """Check a single operation for missing component references."""
+        req_body = operation.get("requestBody", {})
+        content = req_body.get("content", {})
+        multipart = content.get("multipart/form-data", {})
+        schema_ref = multipart.get("schema", {})
+
+        if "$ref" in schema_ref:
+            ref = schema_ref["$ref"]
+            if ref.startswith("#/components/schemas/"):
+                component_name = ref[len("#/components/schemas/") :]
+                
+                if self._is_component_missing(schema_dict, component_name):
+                    missing_refs.append((component_name, ref))
+
+    def _is_component_missing(self, schema_dict: dict, component_name: str) -> bool:
+        """Check if a component is missing from the schema."""
+        components = schema_dict.get("components", {})
+        schemas = components.get("schemas", {})
+        return component_name not in schemas
+
+    def _add_missing_components(self, schema_dict: dict, missing_refs: list[tuple[str, str]]) -> None:
+        """Add missing components to the schema."""
         components = schema_dict.setdefault("components", {})
         schemas = components.setdefault("schemas", {})
 
@@ -98,9 +129,6 @@ def get_openapi_schema(self, **kwargs):
                 "required": ["file"],
             }
 
-        # Rebuild the schema with the added components
-        return schema.__class__(**schema_dict)
-
 
 def create_test_app():
     """Create a test app with the fixed resolver."""