Repository: apache/airflow
File: providers/google/src/airflow/providers/google/cloud/transfers/calendar_to_gcs.py (Python)

Code before:
        self,
        events: list[Any],
    ) -> str:
        gcs_hook = GCSHook(
            gcp_conn_id=self.gcp_conn_id,
            impersonation_chain=self.impersonation_chain,
        )

        # Construct destination file path
        file_name = f"{self.calendar_id}.json".replace(" ", "_")
        dest_file_name = (
            f"{self.destination_path.strip('/')}/{file_name}" if self.destination_path else file_name
        )

        with NamedTemporaryFile("w+") as temp_file:
            # Write data
            json.dump(events, temp_file)
            temp_file.flush()

            # Upload to GCS
            gcs_hook.upload(
                bucket_name=self.destination_bucket,
                object_name=dest_file_name,
                filename=temp_file.name,
            )
        return f"gs://{self.destination_bucket}/{dest_file_name}"

    def execute(self, context) -> str | list[str]:
        calendar_hook = GoogleCalendarHook(
            api_version=self.api_version,
            gcp_conn_id=self.gcp_conn_id,
            impersonation_chain=self.impersonation_chain,
        )

        events = calendar_hook.get_events(
            calendar_id=self.calendar_id,
            i_cal_uid=self.i_cal_uid,
            max_attendees=self.max_attendees,
            max_results=self.max_results,
            order_by=self.order_by,
            private_extended_property=self.private_extended_property,
            q=self.text_search_query,
            shared_extended_property=self.shared_extended_property,
            show_deleted=self.show_deleted,
            show_hidden_invitation=self.show_hidden_invitation,
            single_events=self.single_events,
            sync_token=self.sync_token,
            time_max=self.time_max,
            time_min=self.time_min,
            time_zone=self.time_zone,
            updated_min=self.updated_min,

Review comment (performance):
I went over the PR again and noticed that changing the returned value from `dest_file_name` to the full URI could be a breaking change, so we should deprecate it slowly (similar to the `unwrap_single=False`).
Add another flag `return_gcs_uri` with `False` as default.
Only if `unwrap_single` is also `False`, it should return `dest_file_name`.
`return_gcs_uri` cannot be set to `True` altogether with `unwrap_single=True` (it should raise `ValueError`, as in the future we plan to align all operators with returning list of URIs.
Please handle these case, including unit tests.

Code after:
                stacklevel=2,
            )
        else:
            self.unwrap_single = unwrap_single

        if return_gcs_uri is None:
            self.return_gcs_uri = False
            warnings.warn(
                "The return_gcs_uri parameter is deprecated and will be removed in a future release. "
                "In the future, operators will always return full GCS URIs.",
                FutureWarning,
                stacklevel=2,
            )
        else:
            self.return_gcs_uri = return_gcs_uri

        # Validate that the combination is valid: we cannot have both unwrap_single=False
        # and return_gcs_uri=False as this combination is backward-incompatible with the
        # future behavior of always returning lists of URIs.
        # Note: After None handling above, both values are guaranteed to be bool (not None).
        if not self.unwrap_single and not self.return_gcs_uri:
            raise ValueError(
                "return_gcs_uri cannot be False together with unwrap_single=False. "
                "In the future, all operators will return list of URIs, so please set "
                "return_gcs_uri=True when using unwrap_single=False.",
            )

    def _upload_data(
        self,
        events: list[Any],
    ) -> tuple[str, str]:
        gcs_hook = GCSHook(
            gcp_conn_id=self.gcp_conn_id,
            impersonation_chain=self.impersonation_chain,
        )

        # Construct destination file path
        file_name = f"{self.calendar_id}.json".replace(" ", "_")
        dest_file_name = (
            f"{self.destination_path.strip('/')}/{file_name}" if self.destination_path else file_name
        )

        with NamedTemporaryFile("w+") as temp_file:
            # Write data
            json.dump(events, temp_file)
            temp_file.flush()

            # Upload to GCS
            gcs_hook.upload(
                bucket_name=self.destination_bucket,
                object_name=dest_file_name,