Repository: apache/airflow
File: providers/google/src/airflow/providers/google/marketing_platform/operators/display_video.py (Python)

Code before:
    )

    def __init__(
        self,
        *,
        operation_name: str,
        bucket_name: str,
        object_name: str,
        gzip: bool = False,
        api_version: str = "v4",
        gcp_conn_id: str = "google_cloud_default",
        impersonation_chain: str | Sequence[str] | None = None,
        unwrap_single: bool = True,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)
        self.operation_name = operation_name
        self.bucket_name = bucket_name
        self.object_name = object_name
        self.gzip = gzip
        self.api_version = api_version
        self.gcp_conn_id = gcp_conn_id
        self.impersonation_chain = impersonation_chain
        self.unwrap_single = unwrap_single

    def execute(self, context: Context) -> Union[str, list[str]]:
        hook = GoogleDisplayVideo360Hook(
            gcp_conn_id=self.gcp_conn_id,
            api_version=self.api_version,
            impersonation_chain=self.impersonation_chain,
        )
        gcs_hook = GCSHook(
            gcp_conn_id=self.gcp_conn_id,
            impersonation_chain=self.impersonation_chain,
        )

        self.log.info("Retrieving operation...")
        operation_state = hook.get_sdf_download_operation(operation_name=self.operation_name)

        self.log.info("Creating file for upload...")
        media = hook.download_media(resource_name=operation_state["response"]["resourceName"])

        self.log.info("Sending file to the Google Cloud Storage...")
        with tempfile.TemporaryDirectory() as tmp_dir:
            zip_path = os.path.join(tmp_dir, "sdf.zip")

            # Download the ZIP
            with open(zip_path, "wb") as f:
                hook.download_content_from_request(f, media, chunk_size=1024 * 1024)

            # Extract CSV

Review comment (suggestion):

```suggestion
    def execute(self, context: Context) -> str | list[str]:
```

Code after:

    def __init__(
        self,
        *,
        operation_name: str,
        bucket_name: str,
        object_name: str,
        gzip: bool = False,
        api_version: str = "v4",
        gcp_conn_id: str = "google_cloud_default",
        impersonation_chain: str | Sequence[str] | None = None,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)
        self.operation_name = operation_name
        self.bucket_name = bucket_name
        self.object_name = object_name
        self.gzip = gzip
        self.api_version = api_version
        self.gcp_conn_id = gcp_conn_id
        self.impersonation_chain = impersonation_chain

    def execute(self, context: Context) -> str:
        hook = GoogleDisplayVideo360Hook(
            gcp_conn_id=self.gcp_conn_id,
            api_version=self.api_version,
            impersonation_chain=self.impersonation_chain,
        )
        gcs_hook = GCSHook(
            gcp_conn_id=self.gcp_conn_id,
            impersonation_chain=self.impersonation_chain,
        )

        self.log.info("Retrieving operation...")
        operation_state = hook.get_sdf_download_operation(operation_name=self.operation_name)

        self.log.info("Creating file for upload...")
        media = hook.download_media(resource_name=operation_state["response"]["resourceName"])

        self.log.info("Sending file to the Google Cloud Storage...")
        with tempfile.TemporaryDirectory() as tmp_dir:
            zip_path = os.path.join(tmp_dir, "sdf.zip")

            # Download the ZIP
            with open(zip_path, "wb") as f:
                hook.download_content_from_request(f, media, chunk_size=1024 * 1024)

            # Extract CSV
            with zipfile.ZipFile(zip_path, "r") as zip_ref:
                zip_ref.extractall(tmp_dir)