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

Code before:
            gcp_conn_id=self.gcp_conn_id,
            impersonation_chain=self.impersonation_chain,
        )

        with tempfile.NamedTemporaryFile() as temp_file:
            self.log.info("Downloading data from blob: %s", self.blob_name)
            azure_hook.get_file(
                file_path=temp_file.name,
                container_name=self.container_name,
                blob_name=self.blob_name,
            )
            self.log.info(
                "Uploading data from blob's: %s into GCP bucket: %s", self.object_name, self.bucket_name
            )
            gcs_hook.upload(
                bucket_name=self.bucket_name,
                object_name=self.object_name,
                filename=temp_file.name,
                gzip=self.gzip,
            )
            self.log.info(
                "Resources have been uploaded from blob: %s to GCS bucket:%s",
                self.blob_name,
                self.bucket_name,
            )
        return [f"gs://{self.bucket_name}/{self.object_name}"]

    def get_openlineage_facets_on_start(self):
        from airflow.providers.common.compat.openlineage.facet import Dataset
        from airflow.providers.openlineage.extractors import OperatorLineage

        wasb_hook = WasbHook(wasb_conn_id=self.wasb_conn_id)
        account_name = wasb_hook.get_conn().account_name

        return OperatorLineage(
            inputs=[Dataset(namespace=f"wasbs://{self.container_name}@{account_name}", name=self.blob_name)],
            outputs=[Dataset(namespace=f"gs://{self.bucket_name}", name=self.object_name)],
        )


Review comment (suggestion):
See my comment in the original issue regarding breaking changes (relevant to all related PRs):
https://github.com/apache/airflow/issues/11323#issuecomment-3797323872

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

    template_fields: Sequence[str] = (
        "blob_name",
        "container_name",
        "bucket_name",
        "object_name",
        "filename",
    )

    def execute(self, context: Context) -> str | list[str]:
        azure_hook = WasbHook(wasb_conn_id=self.wasb_conn_id)
        gcs_hook = GCSHook(
            gcp_conn_id=self.gcp_conn_id,
            impersonation_chain=self.impersonation_chain,
        )

        with tempfile.NamedTemporaryFile() as temp_file:
            self.log.info("Downloading data from blob: %s", self.blob_name)
            azure_hook.get_file(
                file_path=temp_file.name,
                container_name=self.container_name,
                blob_name=self.blob_name,
            )
            self.log.info(
                "Uploading data from blob's: %s into GCP bucket: %s", self.object_name, self.bucket_name
            )
            gcs_hook.upload(
                bucket_name=self.bucket_name,
                object_name=self.object_name,
                filename=temp_file.name,
                gzip=self.gzip,
            )
            self.log.info(
                "Resources have been uploaded from blob: %s to GCS bucket:%s",
                self.blob_name,
                self.bucket_name,
            )
        gcs_uri = f"gs://{self.bucket_name}/{self.object_name}"
        if self.unwrap_single:
            return gcs_uri
        return [gcs_uri]

    def get_openlineage_facets_on_start(self):
        from airflow.providers.common.compat.openlineage.facet import Dataset
        from airflow.providers.openlineage.extractors import OperatorLineage