def download_file_from_pipeline_s3()

in src/s3_helper.py [0:0]

• 25 lines of code
• 2 McCabe index (conditional complexity)

def download_file_from_pipeline_s3(job_data, artifact, file_in_zip, download_dir):
    """Pulls artifact credentials from job_data then downloads specific file from the artifact to /tmp

    Args:
        job_data (dict): Job_data section of pipeline event
        artifact (dict): Artifact object from pipeline to pull file from
        file_in_zip (str): File within the artifact dict to download
        download_dir (str): Temporary directory to download file to

    Returns:
        str: Full path to the downloaded file
    """
    logger.debug(f'Getting file ({file_in_zip}) from S3...')

    credentials = {
        'AccessKeyId': job_data['artifactCredentials']['accessKeyId'],
        'SecretAccessKey': job_data['artifactCredentials']['secretAccessKey'],
        'SessionToken': job_data['artifactCredentials']['sessionToken']
    }
    session = boto3_session(credentials=credentials)

    bucket = artifact['location']['s3Location']['bucketName']
    artifact_path = artifact['location']['s3Location']['objectKey']
    zip_file = artifact_path.split('/')[2]

    try:
        logger.debug(f'Downloading {artifact_path} from S3 Bucket ({bucket})...')
        _response = s3_download_file(
            bucket_name=bucket,
            input_file_name=artifact_path,
            output_file_name=f"{download_dir}/{zip_file}",
            session=session
        )
        with zipfile.ZipFile(f"{download_dir}/{zip_file}", "r") as z:
            z.extract(file_in_zip, download_dir)

        return str(f"{download_dir}/{file_in_zip}")

    except (KeyError, AttributeError, OSError) as e:
        logger.error(f'Something went wrong trying to download file. {e}')
        raise Exception(str(e)) from e