Repository: apache/airflow
File: providers/http/src/airflow/providers/http/operators/http.py (Python)

Code before:
        Create a default response maker function based on the type of response.

        :param response: The response object or list of response objects.
        :return: A function that returns response text(s).
        """
        if isinstance(response, Response):
            response_object = response  # Makes mypy happy
            return lambda: response_object.text

        response_list: list[Response] = response  # Makes mypy happy
        return lambda: [entry.text for entry in response_list]

    def execute_complete(
        self, context: Context, event: dict, paginated_responses: None | list[Response] = None
    ):
        """
        Execute callback when the trigger fires; returns immediately.

        Relies on trigger to throw an exception, otherwise it assumes execution was successful.
        """
        if event["status"] == "success":
            try:
                response = HttpResponseSerializer.deserialize(event["response"])
            except TypeError:
                # old pickle format detected - handle it but issue a deprecation warning
                self.log.warning(
                    "Deserializing HTTP response using legacy pickle format. "
                    "This format is deprecated and will be removed in a future version."
                )
                response = pickle.loads(base64.standard_b64decode(event["response"]))

            self.paginate_async(context=context, response=response, previous_responses=paginated_responses)
            return self.process_response(context=context, response=response)
        raise AirflowException(f"Unexpected error in the operation: {event['message']}")

    def paginate_async(
        self, context: Context, response: Response, previous_responses: None | list[Response] = None
    ):
        if self.pagination_function:
            all_responses = previous_responses or []
            all_responses.append(response)

            next_page_params = self.pagination_function(response)
            if not next_page_params:
                return self.process_response(context=context, response=all_responses)
            self.defer(
                trigger=HttpTrigger(
                    http_conn_id=self.http_conn_id,
                    auth_type=serialize_auth_type(self._resolve_auth_type()),
                    method=self.method,
                    **self._merge_next_page_parameters(next_page_params),

Review comment (suggestion):
I think we should not handle it, it does not remove the possibility of deserializing arbitrary value. I think we should add a changelog entry specifying that all Dags using deferrable HTTP request should be paused and completed before upgrading to this provider version.

Code after:
        :param response: The response object or list of response objects.
        :return: A function that returns response text(s).
        """
        if isinstance(response, Response):
            response_object = response  # Makes mypy happy
            return lambda: response_object.text

        response_list: list[Response] = response  # Makes mypy happy
        return lambda: [entry.text for entry in response_list]

    def execute_complete(
        self, context: Context, event: dict, paginated_responses: None | list[Response] = None
    ):
        """
        Execute callback when the trigger fires; returns immediately.

        Relies on trigger to throw an exception, otherwise it assumes execution was successful.
        """
        if event["status"] == "success":
            response = HttpResponseSerializer.deserialize(event["response"])

            self.paginate_async(context=context, response=response, previous_responses=paginated_responses)
            return self.process_response(context=context, response=response)
        raise AirflowException(f"Unexpected error in the operation: {event['message']}")

    def paginate_async(
        self, context: Context, response: Response, previous_responses: None | list[Response] = None
    ):
        if self.pagination_function:
            all_responses = previous_responses or []
            all_responses.append(response)

            next_page_params = self.pagination_function(response)
            if not next_page_params:
                return self.process_response(context=context, response=all_responses)
            self.defer(
                trigger=HttpTrigger(
                    http_conn_id=self.http_conn_id,
                    auth_type=serialize_auth_type(self._resolve_auth_type()),
                    method=self.method,
                    **self._merge_next_page_parameters(next_page_params),
                ),
                method_name="execute_complete",
                kwargs={"paginated_responses": all_responses},
            )

    def _merge_next_page_parameters(self, next_page_params: dict) -> dict:
        """
        Merge initial request parameters with next page parameters.

        Merge initial requests parameters with the ones for the next page, generated by