Repository: spotify/luigi
File: luigi/contrib/batch.py (Python)

Code before:
        """
        response = self._client.describe_jobs(jobs=[job_id])

        # Error checking
        status_code = response['ResponseMetadata']['HTTPStatusCode']
        if status_code != 200:
            msg = 'Job status request received status code {0}:\n{1}'
            raise Exception(msg.format(status_code, response))

        return response['jobs'][0]['status']

    def get_logs(self, log_stream_name, get_last=50):
        """Retrieve log stream from CloudWatch"""
        response = self._log_client.get_log_events(
            logGroupName='/aws/batch/job', 
            logStreamName=log_stream_name,
            startFromHead=False)
        events = response['events']
        return '\n'.join(e['message'] for e in events[-get_last:])

    def submit_job(self, job_definition, parameters, job_name=None, queue=None):
        """Wrap submit_job with useful defaults"""
        if job_name is None:
            job_name = _random_id()
        response = self._client.submit_job(
            jobName = job_name,
            jobQueue = queue or self.get_active_queue(),
            jobDefinition = job_definition,
            parameters = parameters
        )
        return response['jobId']

    def wait_on_job(self, job_id):
        """Poll task status until STOPPED"""
        
        while True:
            status = self.get_job_status(job_id)
            if status == 'SUCCEEDED':
                logger.info('Batch job {} SUCCEEDED'.format(job_id))
                return True
            elif status == 'FAILED':
                # Raise and notify if job failed
                jobs = self._client.describe_jobs(jobs=[job_id])['jobs']
                job_str = json.dumps(jobs, indent=4)
                logger.debug('Job details:\n' + job_str)
                
                log_stream_name = jobs[0]['attempts'][0]['container']['logStreamName']
                logs = self.get_logs(log_stream_name)
                raise BatchJobException('Job {} failed: {}'.format(
                    job_id, logs))


Review comment (suggestion):
Remove the explicit conditional check for if `job_name is None` by updating this to `jobName=job_name or _random_id(),`

Code after:
        """
        response = self._client.describe_jobs(jobs=[job_id])

        # Error checking
        status_code = response['ResponseMetadata']['HTTPStatusCode']
        if status_code != 200:
            msg = 'Job status request received status code {0}:\n{1}'
            raise Exception(msg.format(status_code, response))

        return response['jobs'][0]['status']

    def get_logs(self, log_stream_name, get_last=50):
        """Retrieve log stream from CloudWatch"""
        response = self._log_client.get_log_events(
            logGroupName='/aws/batch/job',
            logStreamName=log_stream_name,
            startFromHead=False)
        events = response['events']
        return '\n'.join(e['message'] for e in events[-get_last:])

    def submit_job(self, job_definition, parameters, job_name=None, queue=None):
        """Wrap submit_job with useful defaults"""
        if job_name is None:
            job_name = _random_id()
        response = self._client.submit_job(
            jobName=job_name,
            jobQueue=queue or self.get_active_queue(),
            jobDefinition=job_definition,
            parameters=parameters
        )
        return response['jobId']

    def wait_on_job(self, job_id):
        """Poll task status until STOPPED"""

        while True:
            status = self.get_job_status(job_id)
            if status == 'SUCCEEDED':
                logger.info('Batch job {} SUCCEEDED'.format(job_id))
                return True
            elif status == 'FAILED':
                # Raise and notify if job failed
                jobs = self._client.describe_jobs(jobs=[job_id])['jobs']
                job_str = json.dumps(jobs, indent=4)
                logger.debug('Job details:\n' + job_str)

                log_stream_name = jobs[0]['attempts'][0]['container']['logStreamName']
                logs = self.get_logs(log_stream_name)
                raise BatchJobException('Job {} failed: {}'.format(
                    job_id, logs))