Function import_api_from_openapi_spec has a Cognitive Complexity of 158 (exceeds 5 allowed). Consider refactoring.

Open

def import_api_from_openapi_spec(
    rest_api: RestAPI, body: dict, context: RequestContext
) -> Optional[RestAPI]:
    """Import an API from an OpenAPI spec document"""

• Read up
  Read up

File helpers.py has 1197 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import contextlib
import copy
import hashlib
import json
import logging

Function apply_json_patch_safe has a Cognitive Complexity of 38 (exceeds 5 allowed). Consider refactoring.

Open

def apply_json_patch_safe(subject, patch_operations, in_place=True, return_list=False):
    """Apply JSONPatch operations, using some customizations for compatibility with API GW
    resources."""

    results = []

• Read up
  Read up

Function resolve_model has a Cognitive Complexity of 30 (exceeds 5 allowed). Consider refactoring.

Open

    def resolve_model(self, model: dict) -> dict | None:
        resolved_model = copy.deepcopy(model)
        model_names = set()

        def _look_for_ref(sub_model):

• Read up
  Read up

Function get_resource_for_path has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def get_resource_for_path(
    path: str, method: str, path_map: Dict[str, Dict]
) -> tuple[Optional[str], Optional[dict]]:
    matches = []
    # creates a regex from the input path if there are parameters, e.g /foo/{bar}/baz -> /foo/[

• Read up
  Read up

Function log_template has 11 arguments (exceeds 4 allowed). Consider refactoring.

Open

def log_template(

Function _resolve_refpath has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def _resolve_refpath(self, refpath: str) -> dict:
        if refpath in self._refpaths and not self.allow_recursive:
            raise Exception("recursion detected with allow_recursive=False")

        # We don't resolve the Model definition, we will return a absolute reference to the model like AWS

• Read up
  Read up

Function resolve has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def resolve(self, context: ApiInvocationContext) -> IntegrationParameters:
        """
        Resolve method request parameters into integration request parameters.
        Integration request parameters, in the form of path variables, query strings
        or headers, can be mapped from any defined method request parameters

• Read up
  Read up

Function method_request_dict has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def method_request_dict(self, context: ApiInvocationContext) -> Dict[str, Any]:
        """
        Build a dict with all method request parameters and their values.
        :return: dict with all method request parameters and their values,
        and all keys in lowercase

• Read up
  Read up

Function select_integration_response has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def select_integration_response(matched_part: str, invocation_context: ApiInvocationContext):
    int_responses = invocation_context.integration.get("integrationResponses") or {}
    if select_by_pattern := [
        response
        for response in int_responses.values()

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if content_type == APPLICATION_JSON:
                            model_ref = media_type.get("schema", {}).get("$ref")
                            continue
                        LOG.warning(

Avoid deeply nested control flow statements.

Open

                        if param_location == "query":
                            param_location = "querystring"

Function connect_api_gateway_to_sqs has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def connect_api_gateway_to_sqs(gateway_name, stage_name, queue_arn, path, account_id, region_name):

Function _resolve_references has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _resolve_references(self, data) -> Union[dict, list]:
        if self._is_ref(data):
            return self._resolve_refpath(data["$ref"])

        if isinstance(data, dict):

• Read up
  Read up

Function _remove_self_ref has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _remove_self_ref(self, resolved_schema: dict):
        for key, value in resolved_schema.items():
            if key == "$ref":
                ref_name = value.rsplit("/", maxsplit=1)[-1]
                if ref_name == self.model_name:

• Read up
  Read up

Function resolve has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def resolve(self, context: ApiInvocationContext) -> Dict[str, str]:
        """
        Resolve integration response parameters into method response parameters.
        Integration response parameters can map header, body,
        or static values to the header type of the method response.

• Read up
  Read up

Function get_resolved_model has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def get_resolved_model(self) -> dict | None:
        if not (resolved_model := self.rest_api_container.resolved_models.get(self.model_name)):
            model = self.rest_api_container.models.get(self.model_name)
            if not model:
                return None

• Read up
  Read up

Avoid too many return statements within this function.

Open

        return sorted_matches[0]

Avoid too many return statements within this function.

Open

    return None, None