Repository: apache/airflow
File: providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py (Python)

Code before:
    ):
        super().__init__(**kwargs)

        self.source_bucket_key = source_bucket_key
        self.dest_bucket_key = dest_bucket_key
        self.source_bucket_name = source_bucket_name
        self.dest_bucket_name = dest_bucket_name
        self.source_version_id = source_version_id
        self.acl_policy = acl_policy
        self.meta_data_directive = meta_data_directive
        self.kms_key_id = kms_key_id
        self.kms_encryption_type = kms_encryption_type

    def execute(self, context: Context):
        copy_args = {}
        if self.kms_key_id:
            copy_args["SSEKMSKeyId"] = self.kms_key_id
        if self.kms_encryption_type:
            copy_args["ServerSideEncryption"] = self.kms_encryption_type

        self.hook.copy_object(
            self.source_bucket_key,
            self.dest_bucket_key,
            self.source_bucket_name,
            self.dest_bucket_name,
            self.source_version_id,
            self.acl_policy,
            self.meta_data_directive,
            **copy_args,
        )

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

        dest_bucket_name, dest_bucket_key = self.hook.get_s3_bucket_key(
            self.dest_bucket_name, self.dest_bucket_key, "dest_bucket_name", "dest_bucket_key"
        )

        source_bucket_name, source_bucket_key = self.hook.get_s3_bucket_key(
            self.source_bucket_name, self.source_bucket_key, "source_bucket_name", "source_bucket_key"
        )

        input_dataset = Dataset(
            namespace=f"s3://{source_bucket_name}",
            name=source_bucket_key,
        )
        output_dataset = Dataset(
            namespace=f"s3://{dest_bucket_name}",
            name=dest_bucket_key,
        )

Review comment (refactor):
Passing the KMS parameters via **kwargs makes the new functionality dependent on copy_object continuing to forward those parameters. The tests you added do help protect against refactors of copy_object breaking this behavior, but the contract itself isn’t explicitly represented in the public API. I understand the motivation to avoid modifying the public API, but that argument is more defensible for breaking changes. Adding optional parameters would be non-breaking here and would make the contract more explicit. This is of course just a suggestion as your tests prevent regressions. 

Code after:
    ):
        super().__init__(**kwargs)

        self.source_bucket_key = source_bucket_key
        self.dest_bucket_key = dest_bucket_key
        self.source_bucket_name = source_bucket_name
        self.dest_bucket_name = dest_bucket_name
        self.source_version_id = source_version_id
        self.acl_policy = acl_policy
        self.meta_data_directive = meta_data_directive
        self.kms_key_id = kms_key_id
        self.kms_encryption_type = kms_encryption_type

    def execute(self, context: Context):
        self.hook.copy_object(
            self.source_bucket_key,
            self.dest_bucket_key,
            self.source_bucket_name,
            self.dest_bucket_name,
            self.source_version_id,
            self.acl_policy,
            self.meta_data_directive,
            self.kms_key_id,
            self.kms_encryption_type,
        )

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

        dest_bucket_name, dest_bucket_key = self.hook.get_s3_bucket_key(
            self.dest_bucket_name, self.dest_bucket_key, "dest_bucket_name", "dest_bucket_key"
        )

        source_bucket_name, source_bucket_key = self.hook.get_s3_bucket_key(
            self.source_bucket_name, self.source_bucket_key, "source_bucket_name", "source_bucket_key"
        )

        input_dataset = Dataset(
            namespace=f"s3://{source_bucket_name}",
            name=source_bucket_key,
        )
        output_dataset = Dataset(
            namespace=f"s3://{dest_bucket_name}",
            name=dest_bucket_key,
        )

        return OperatorLineage(
            inputs=[input_dataset],
            outputs=[output_dataset],
        )