def copy_object()

in CommonLayerCode/datalake-library/python/datalake_library/interfaces/s3_interface.py [0:0]

• 28 lines of code
• 6 McCabe index (conditional complexity)

    def copy_object(self, source_bucket, source_key, dest_bucket, dest_key=None, kms_key=None):
        source_key = unquote_plus(source_key)
        self._logger.info("Copying object {}/{} to {}/{}".format(source_bucket,
                                                                 source_key,
                                                                 dest_bucket,
                                                                 dest_key if dest_key else source_key))
        try:
            extra_kwargs = {}
            if kms_key:
                extra_kwargs = {
                    "ServerSideEncryption": "aws:kms",
                    "SSEKMSKeyId": kms_key
                }
            copy_source = {
                'Bucket': source_bucket,
                'Key': source_key
            }
            self._s3_resource.meta.client.copy(copy_source,
                                               dest_bucket,
                                               dest_key if dest_key else source_key,
                                               ExtraArgs=extra_kwargs)
        except ClientError:
            msg = 'Error copying object: {}/{} to {}/{}'.format(source_bucket,
                                                                source_key,
                                                                dest_bucket,
                                                                dest_key if dest_key else source_key)
            self._logger.exception(msg)
            raise