refactor: reduce cognitive complexity in OpenAPI schema generation

- Refactored get_openapi_schema method to reduce cognitive complexity from 27 to 15
  - Split into 7 helper methods for better maintainability
  - Enhanced error handling and code organization

- Refactored find_missing_component_references to reduce complexity from 23 to 15
  - Split into 5 helper methods with proper separation of concerns
  - Fixed null pointer bug in _get_existing_schemas function

- Reorganized examples directory structure
  - Moved upload file examples to examples/event_handler/
  - Updated documentation and imports accordingly

- All OpenAPI tests passing (220/220)

Author: oyiz-michael

aws_lambda_powertools/event_handler/api_gateway.py

@@ -1774,72 +1774,114 @@ def get_openapi_schema(
         OpenAPI: pydantic model
             The OpenAPI schema as a pydantic model.
         """
+        # Resolve configuration with fallbacks to openapi_config
+        config = self._resolve_openapi_config(
+            title=title,
+            version=version,
+            openapi_version=openapi_version,
+            summary=summary,
+            description=description,
+            tags=tags,
+            servers=servers,
+            terms_of_service=terms_of_service,
+            contact=contact,
+            license_info=license_info,
+            security_schemes=security_schemes,
+            security=security,
+            external_documentation=external_documentation,
+            openapi_extensions=openapi_extensions,
+        )
 
-        # DEPRECATION: Will be removed in v4.0.0. Use configure_api() instead.
-        # Maintained for backwards compatibility.
-        # See: https://github.com/aws-powertools/powertools-lambda-python/issues/6122
-        if title == DEFAULT_OPENAPI_TITLE and self.openapi_config.title:
-            title = self.openapi_config.title
-
-        if version == DEFAULT_API_VERSION and self.openapi_config.version:
-            version = self.openapi_config.version
-
-        if openapi_version == DEFAULT_OPENAPI_VERSION and self.openapi_config.openapi_version:
-            openapi_version = self.openapi_config.openapi_version
-
-        summary = summary or self.openapi_config.summary
-        description = description or self.openapi_config.description
-        tags = tags or self.openapi_config.tags
-        servers = servers or self.openapi_config.servers
-        terms_of_service = terms_of_service or self.openapi_config.terms_of_service
-        contact = contact or self.openapi_config.contact
-        license_info = license_info or self.openapi_config.license_info
-        security_schemes = security_schemes or self.openapi_config.security_schemes
-        security = security or self.openapi_config.security
-        external_documentation = external_documentation or self.openapi_config.external_documentation
-        openapi_extensions = openapi_extensions or self.openapi_config.openapi_extensions
+        # Build base OpenAPI structure
+        output = self._build_base_openapi_structure(config)
 
-        from pydantic.json_schema import GenerateJsonSchema
+        # Process routes and build paths/components
+        paths, definitions = self._process_routes_for_openapi(config["security_schemes"])
 
-        from aws_lambda_powertools.event_handler.openapi.compat import (
-            get_compat_model_name_map,
-            get_definitions,
-        )
-        from aws_lambda_powertools.event_handler.openapi.models import OpenAPI, PathItem, Tag
-        from aws_lambda_powertools.event_handler.openapi.types import (
-            COMPONENT_REF_TEMPLATE,
-        )
+        # Build final components and paths
+        components = self._build_openapi_components(definitions, config["security_schemes"])
+        output.update(self._finalize_openapi_output(components, config["tags"], paths, config["external_documentation"]))
 
-        openapi_version = self._determine_openapi_version(openapi_version)
+        # Apply schema fixes and return result
+        return self._apply_schema_fixes(output)
+
+    def _resolve_openapi_config(self, **kwargs) -> dict[str, Any]:
+        """Resolve OpenAPI configuration with fallbacks to openapi_config."""
+        # DEPRECATION: Will be removed in v4.0.0. Use configure_api() instead.
+        # Maintained for backwards compatibility.
+        # See: https://github.com/aws-powertools/powertools-lambda-python/issues/6122
+        resolved = {}
+        
+        # Handle title with fallback
+        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
+        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
+        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
+        resolved.update({
+            "summary": kwargs["summary"] or self.openapi_config.summary,
+            "description": kwargs["description"] or self.openapi_config.description,
+            "tags": kwargs["tags"] or self.openapi_config.tags,
+            "servers": kwargs["servers"] or self.openapi_config.servers,
+            "terms_of_service": kwargs["terms_of_service"] or self.openapi_config.terms_of_service,
+            "contact": kwargs["contact"] or self.openapi_config.contact,
+            "license_info": kwargs["license_info"] or self.openapi_config.license_info,
+            "security_schemes": kwargs["security_schemes"] or self.openapi_config.security_schemes,
+            "security": kwargs["security"] or self.openapi_config.security,
+            "external_documentation": kwargs["external_documentation"] or self.openapi_config.external_documentation,
+            "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"])
 
         # Start with the bare minimum required for a valid OpenAPI schema
-        info: dict[str, Any] = {"title": title, "version": version}
+        info: dict[str, Any] = {"title": config["title"], "version": config["version"]}
 
         optional_fields = {
-            "summary": summary,
-            "description": description,
-            "termsOfService": terms_of_service,
-            "contact": contact,
-            "license": license_info,
+            "summary": config["summary"],
+            "description": config["description"],
+            "termsOfService": config["terms_of_service"],
+            "contact": config["contact"],
+            "license": config["license_info"],
         }
 
         info.update({field: value for field, value in optional_fields.items() if value})
 
+        openapi_extensions = config["openapi_extensions"]
         if not isinstance(openapi_extensions, dict):
             openapi_extensions = {}
 
-        output: dict[str, Any] = {
+        return {
             "openapi": openapi_version,
             "info": info,
-            "servers": self._get_openapi_servers(servers),
-            "security": self._get_openapi_security(security, security_schemes),
+            "servers": self._get_openapi_servers(config["servers"]),
+            "security": self._get_openapi_security(config["security"], config["security_schemes"]),
             **openapi_extensions,
         }
 
-        if external_documentation:
-            output["externalDocs"] = external_documentation
+    def _process_routes_for_openapi(self, security_schemes: dict[str, SecurityScheme] | None) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]:
+        """Process all routes and build paths and definitions."""
+        from pydantic.json_schema import GenerateJsonSchema
+        from aws_lambda_powertools.event_handler.openapi.compat import (
+            get_compat_model_name_map,
+            get_definitions,
+        )
+        from aws_lambda_powertools.event_handler.openapi.types import COMPONENT_REF_TEMPLATE
 
-        components: dict[str, dict[str, Any]] = {}
         paths: dict[str, dict[str, Any]] = {}
         operation_ids: set[str] = set()
 
@@ -1857,15 +1899,8 @@ def get_openapi_schema(
 
         # Add routes to the OpenAPI schema
         for route in all_routes:
-            if route.security and not _validate_openapi_security_parameters(
-                security=route.security,
-                security_schemes=security_schemes,
-            ):
-                raise SchemaValidationError(
-                    "Security configuration was not found in security_schemas or security_schema was not defined. "
-                    "See: https://docs.powertools.aws.dev/lambda/python/latest/core/event_handler/api_gateway/#security-schemes",
-                )
-
+            self._validate_route_security(route, security_schemes)
+            
             if not route.include_in_schema:
                 continue
 
@@ -1883,19 +1918,50 @@ def get_openapi_schema(
                 if path_definitions:
                     definitions.update(path_definitions)
 
+        return paths, definitions
+
+    def _validate_route_security(self, route, security_schemes: dict[str, SecurityScheme] | None) -> None:
+        """Validate route security configuration."""
+        if route.security and not _validate_openapi_security_parameters(
+            security=route.security,
+            security_schemes=security_schemes,
+        ):
+            raise SchemaValidationError(
+                "Security configuration was not found in security_schemas or security_schema was not defined. "
+                "See: https://docs.powertools.aws.dev/lambda/python/latest/core/event_handler/api_gateway/#security-schemes",
+            )
+
+    def _build_openapi_components(self, definitions: dict[str, dict[str, Any]], security_schemes: dict[str, SecurityScheme] | None) -> dict[str, dict[str, Any]]:
+        """Build the components section of the OpenAPI schema."""
+        components: dict[str, dict[str, Any]] = {}
+        
         if definitions:
             components["schemas"] = self._generate_schemas(definitions)
         if security_schemes:
             components["securitySchemes"] = security_schemes
+            
+        return components
+
+    def _finalize_openapi_output(self, components: dict[str, dict[str, Any]], tags, paths: dict[str, dict[str, Any]], external_documentation) -> dict[str, Any]:
+        """Finalize the OpenAPI output with components, tags, and paths."""
+        from aws_lambda_powertools.event_handler.openapi.models import PathItem, Tag
+
+        output = {}
+        
         if components:
             output["components"] = components
         if tags:
             output["tags"] = [Tag(name=tag) if isinstance(tag, str) else tag for tag in tags]
+        if external_documentation:
+            output["externalDocs"] = external_documentation
 
         output["paths"] = {k: PathItem(**v) for k, v in paths.items()}
+        
+        return output
 
-        # Apply patches to fix any issues with the OpenAPI schema
-        # Import here to avoid circular imports
+    def _apply_schema_fixes(self, output: dict[str, Any]) -> OpenAPI:
+        """Apply schema fixes and return the final OpenAPI model."""
+        from aws_lambda_powertools.event_handler.openapi.models import OpenAPI
         from aws_lambda_powertools.event_handler.openapi.upload_file_fix import fix_upload_file_schema
 
         # First create the OpenAPI model
@@ -1906,9 +1972,7 @@ def get_openapi_schema(
         fixed_dict = fix_upload_file_schema(result_dict)
 
         # Reconstruct the model with the fixed dict
-        result = OpenAPI(**fixed_dict)
-
-        return result
+        return OpenAPI(**fixed_dict)
 
     @staticmethod
     def _get_openapi_servers(servers: list[Server] | None) -> list[Server]:

aws_lambda_powertools/event_handler/openapi/upload_file_fix.py

@@ -62,43 +62,86 @@ def find_missing_component_references(schema_dict: dict[str, Any]) -> list[tuple
         A list of tuples containing (reference_name, path_url)
     """
     paths = schema_dict.get("paths", {})
+    existing_schemas = _get_existing_schemas(schema_dict)
     missing_components: list[tuple[str, str]] = []
 
-    # Find all referenced component names that don't exist in the schema
     for path_url, path_item in paths.items():
         if not isinstance(path_item, dict):
             continue
+        _check_path_for_missing_components(path_item, path_url, existing_schemas, missing_components)
 
-        for _method, operation in path_item.items():
-            if not isinstance(operation, dict):
-                continue
+    return missing_components
 
-            if "requestBody" not in operation or not operation["requestBody"]:
-                continue
 
-            request_body = operation["requestBody"]
-            if "content" not in request_body or not request_body["content"]:
-                continue
+def _get_existing_schemas(schema_dict: dict[str, Any]) -> set[str]:
+    """Get the set of existing schema component names."""
+    components = schema_dict.get("components")
+    if components is None:
+        return set()
+    
+    schemas = components.get("schemas")
+    if schemas is None:
+        return set()
+        
+    return set(schemas.keys())
 
-            content = request_body["content"]
-            if "multipart/form-data" not in content:
-                continue
 
-            multipart = content["multipart/form-data"]
+def _check_path_for_missing_components(
+    path_item: dict[str, Any], 
+    path_url: str, 
+    existing_schemas: set[str], 
+    missing_components: 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):
+            continue
+        _check_operation_for_missing_components(operation, path_url, existing_schemas, missing_components)
 
-            # Get schema reference - could be in schema or schema_ (Pydantic v1/v2 difference)
-            schema_ref = get_schema_ref(multipart)
 
-            if schema_ref and isinstance(schema_ref, str) and schema_ref.startswith("#/components/schemas/"):
-                ref_name = schema_ref[len("#/components/schemas/") :]
-                # Check if this component exists
-                components = schema_dict.get("components", {})
-                schemas = components.get("schemas", {})
+def _check_operation_for_missing_components(
+    operation: dict[str, Any], 
+    path_url: str, 
+    existing_schemas: set[str], 
+    missing_components: list[tuple[str, str]]
+) -> None:
+    """Check a single operation for missing component references."""
+    multipart_schema = _extract_multipart_schema(operation)
+    if not multipart_schema:
+        return
+        
+    schema_ref = get_schema_ref(multipart_schema)
+    ref_name = _extract_component_name(schema_ref)
+    
+    if ref_name and ref_name not in existing_schemas:
+        missing_components.append((ref_name, path_url))
 
-                if ref_name not in schemas:
-                    missing_components.append((ref_name, path_url))
 
-    return missing_components
+def _extract_multipart_schema(operation: dict[str, Any]) -> dict[str, Any] | None:
+    """Extract multipart/form-data schema from operation, if it exists."""
+    if "requestBody" not in operation or not operation["requestBody"]:
+        return None
+        
+    request_body = operation["requestBody"]
+    if "content" not in request_body or not request_body["content"]:
+        return None
+        
+    content = request_body["content"]
+    if "multipart/form-data" not in content:
+        return None
+        
+    return content["multipart/form-data"]
+
+
+def _extract_component_name(schema_ref: str | None) -> str | None:
+    """Extract component name from schema reference."""
+    if not schema_ref or not isinstance(schema_ref, str):
+        return None
+        
+    if not schema_ref.startswith("#/components/schemas/"):
+        return None
+        
+    return schema_ref[len("#/components/schemas/"):]
 
 
 def get_schema_ref(multipart: dict[str, Any]) -> str | None: