def _call()

in src/rpdk/core/contract/resource_client.py [0:0]
58 lines of code
5 McCabe index (conditional complexity)

    def _call(self, payload):
        request_without_write_properties = prune_properties(
            payload["requestData"]["resourceProperties"], self.write_only_paths
        )

        previous_request_without_write_properties = None
        if payload["requestData"]["previousResourceProperties"]:
            previous_request_without_write_properties = prune_properties(
                payload["requestData"]["previousResourceProperties"],
                self.write_only_paths,
            )
        payload_to_log = {
            "callbackContext": payload["callbackContext"],
            "action": payload["action"],
            "requestData": {
                "resourceProperties": request_without_write_properties,
                "previousResourceProperties": previous_request_without_write_properties,
                "logicalResourceId": payload["requestData"]["logicalResourceId"],
            },
            "region": payload["region"],
            "awsAccountId": payload["awsAccountId"],
            "bearerToken": payload["bearerToken"],
        }
        LOG.debug(
            "Sending request\n%s",
            json.dumps(payload_to_log, ensure_ascii=False, indent=2),
        )
        payload = json.dumps(payload, ensure_ascii=False, indent=2)
        if self._docker_image:
            if not self._executable_entrypoint:
                raise InvalidProjectError(
                    "executableEntrypoint not set in .rpdk-config. "
                    "Have you run cfn generate?"
                )
            result = (
                self._docker_client.containers.run(
                    self._docker_image,
                    self._executable_entrypoint + " '" + payload + "'",
                )
                .decode()
                .strip()
            )
            LOG.debug("=== Handler execution logs ===")
            LOG.debug(result)
            # pylint: disable=W1401
            regex = "__CFN_RESOURCE_START_RESPONSE__([\s\S]*)__CFN_RESOURCE_END_RESPONSE__"  # noqa: W605,B950 # pylint: disable=C0301
            payload = json.loads(re.search(regex, result).group(1))
        else:
            result = self._client.invoke(
                FunctionName=self._function_name, Payload=payload.encode("utf-8")
            )

            try:
                payload = json.load(result["Payload"])
            except json.decoder.JSONDecodeError as json_error:
                LOG.debug("Received invalid response\n%s", result["Payload"])
                raise ValueError(
                    "Handler Output is not a valid JSON document"
                ) from json_error
        LOG.debug("Received response\n%s", payload)
        return payload