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

Code before:
                replace=False,
                azure_data_lake_conn_id='azure_data_lake_default',
                gcp_conn_id='google_cloud_default'
            )
    """

    template_fields: Sequence[str] = (
        "src_adls",
        "dest_gcs",
        "google_impersonation_chain",
    )
    ui_color = "#f0eee4"

    def __init__(
        self,
        *,
        src_adls: str,
        dest_gcs: str,
        azure_data_lake_conn_id: str,
        gcp_conn_id: str = "google_cloud_default",
        replace: bool = False,
        gzip: bool = False,
        google_impersonation_chain: str | Sequence[str] | None = None,
        **kwargs,
    ) -> None:
        file_system_name = kwargs.pop("file_system_name", None)
        if file_system_name is None:
            raise TypeError(
                "The 'file_system_name' parameter is required. "
                "ADLSListOperator has been migrated from Azure Data Lake Storage Gen1 (retired) "
                "to Gen2, which requires specifying a file system name. "
                "Please add file_system_name='your-container-name' to your operator instantiation."
            )

        super().__init__(
            file_system_name=file_system_name,
            path=src_adls,
            azure_data_lake_conn_id=azure_data_lake_conn_id,
            **kwargs,
        )

        self.src_adls = src_adls
        self.dest_gcs = dest_gcs
        self.replace = replace
        self.gcp_conn_id = gcp_conn_id
        self.gzip = gzip
        self.google_impersonation_chain = google_impersonation_chain

    def execute(self, context: Context):
        # use the super to list all files in an Azure Data Lake path
        files = super().execute(context)

Review comment (suggestion):
@shahar1 Wouldn't it be better to have this logic in a (static) method validate_kwargs which we unpack in constructor like so we don't pollute the constuctor with code before invoking to super one?
```python
super().__init__(
            file_system_name=file_system_name,
            path=src_adls,
            azure_data_lake_conn_id=azure_data_lake_conn_id,
            **validate_kwargs(kwargs),
        )
```

Code after:
                replace=False,
                azure_data_lake_conn_id='azure_data_lake_default',
                gcp_conn_id='google_cloud_default'
            )
    """

    template_fields: Sequence[str] = (
        "src_adls",
        "dest_gcs",
        "google_impersonation_chain",
    )
    ui_color = "#f0eee4"

    def __init__(
        self,
        *,
        src_adls: str,
        dest_gcs: str,
        azure_data_lake_conn_id: str,
        gcp_conn_id: str = "google_cloud_default",
        replace: bool = False,
        gzip: bool = False,
        google_impersonation_chain: str | Sequence[str] | None = None,
        **kwargs,
    ) -> None:
        super().__init__(
            path=src_adls,
            azure_data_lake_conn_id=azure_data_lake_conn_id,
            **self._validate_kwargs(kwargs),
        )

        self.src_adls = src_adls
        self.dest_gcs = dest_gcs
        self.replace = replace
        self.gcp_conn_id = gcp_conn_id
        self.gzip = gzip
        self.google_impersonation_chain = google_impersonation_chain

    @staticmethod
    def _validate_kwargs(kwargs: dict) -> dict:
        file_system_name = kwargs.pop("file_system_name", None)
        if file_system_name is None:
            raise TypeError(
                "The 'file_system_name' parameter is required. "
                "ADLSListOperator has been migrated from Azure Data Lake Storage Gen1 (retired) "
                "to Gen2, which requires specifying a file system name. "
                "Please add file_system_name='your-container-name' to your operator instantiation."
            )
        return {"file_system_name": file_system_name, **kwargs}

    def execute(self, context: Context):