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

Code before:
            as `source_bucket_key`.
        :param source_bucket_name: Name of the S3 bucket where the source object is in.

            It should be omitted when `source_bucket_key` is provided as a full s3:// url.
        :param dest_bucket_name: Name of the S3 bucket to where the object is copied.

            It should be omitted when `dest_bucket_key` is provided as a full s3:// url.
        :param source_version_id: Version ID of the source object (OPTIONAL)
        :param acl_policy: The string to specify the canned ACL policy for the
            object to be copied which is private by default.
        :param meta_data_directive: Whether to `COPY` the metadata from the source object or `REPLACE` it
            with metadata that's provided in the request.
        :param kms_key_id: The ARN, id or alias of the AWS KMS key to use for encrypting the destination object.
            Required if using KMS-based server-side encryption with a non-default key.
        :param kms_encryption_type: Type of KMS encryption to use for the object.
            Can be either "aws:kms" (standard KMS) or "aws:kms:dsse" (double-shielded KMS).
        """
        acl_policy = acl_policy or "private"
        if acl_policy != NO_ACL:
            kwargs["ACL"] = acl_policy
        if meta_data_directive:
            kwargs["MetadataDirective"] = meta_data_directive
        if self._requester_pays:
            kwargs["RequestPayer"] = "requester"

        if (kms_key_id and not kms_encryption_type) or (kms_encryption_type and not kms_key_id):
            message = "kms_key_id and kms_encryption_type must both be specified. Only one was provided."
            self.log.error(message)
            raise ValueError(message)
        if kms_key_id and kms_encryption_type:
            kwargs["SSEKMSKeyId"] = kms_key_id
            kwargs["ServerSideEncryption"] = kms_encryption_type

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

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

        copy_source = {
            "Bucket": source_bucket_name,
            "Key": source_bucket_key,
            "VersionId": source_version_id,
        }
        response = self.get_conn().copy_object(
            Bucket=dest_bucket_name,

Review comment (suggestion):
Nit: this might be slightly less cluttered:

```suggestion
        if bool(kms_key_id) != bool(kms_encryption_type):
```

Basically we just care that they both resolve to the same "something", either a str value each (so truthy) or None (falsey)

Code after:
            as `source_bucket_key`.
        :param source_bucket_name: Name of the S3 bucket where the source object is in.

            It should be omitted when `source_bucket_key` is provided as a full s3:// url.
        :param dest_bucket_name: Name of the S3 bucket to where the object is copied.

            It should be omitted when `dest_bucket_key` is provided as a full s3:// url.
        :param source_version_id: Version ID of the source object (OPTIONAL)
        :param acl_policy: The string to specify the canned ACL policy for the
            object to be copied which is private by default.
        :param meta_data_directive: Whether to `COPY` the metadata from the source object or `REPLACE` it
            with metadata that's provided in the request.
        :param kms_key_id: The ARN, id or alias of the AWS KMS key to use for encrypting the destination object.
            Required if using KMS-based server-side encryption with a non-default key.
        :param kms_encryption_type: Type of KMS encryption to use for the object.
            Can be either "aws:kms" (standard KMS) or "aws:kms:dsse" (double-shielded KMS).
        """
        acl_policy = acl_policy or "private"
        if acl_policy != NO_ACL:
            kwargs["ACL"] = acl_policy
        if meta_data_directive:
            kwargs["MetadataDirective"] = meta_data_directive
        if self._requester_pays:
            kwargs["RequestPayer"] = "requester"

        if bool(kms_key_id) != bool(kms_encryption_type):
            message = "kms_key_id and kms_encryption_type must both be specified. Only one was provided."
            raise ValueError(message)
        if kms_key_id and kms_encryption_type:
            kwargs["SSEKMSKeyId"] = kms_key_id
            kwargs["ServerSideEncryption"] = kms_encryption_type

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

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

        copy_source = {
            "Bucket": source_bucket_name,
            "Key": source_bucket_key,
            "VersionId": source_version_id,
        }
        response = self.get_conn().copy_object(
            Bucket=dest_bucket_name,
            Key=dest_bucket_key,