def get_job_detail()

in core/src/klio_core/dataflow.py [0:0]

• 26 lines of code
• 3 McCabe index (conditional complexity)

    def get_job_detail(self, job_name, gcp_project, region=None):
        """Get verbose job detail given a job name.

        Args:
            job_name (str): Name of Dataflow job.
            gcp_project (str): GCP project in which to search.
            region (str): Region in which to search. Defaults to
                searching all regions in
                :attr:`klio_core.variables.DATAFLOW_REGIONS`.
        Returns:
            dict or None: If found, ``dict`` of detailed job results.
            Otherwise, ``None``.
        """
        basic_job = self.find_job_by_name(job_name, gcp_project, region)
        if not basic_job:
            return None

        job_id = basic_job["id"]
        job_location = basic_job["location"]

        request = (
            self.client.projects()
            .locations()
            .jobs()
            .get(
                projectId=gcp_project,
                location=job_location,
                jobId=job_id,
                view="JOB_VIEW_ALL",
            )
        )
        try:
            response = request.execute()
        # general catch all since the handling would be the same no matter
        # of the exception
        except Exception as e:
            self.logger.warning(
                "Error getting job detail for '%s' in project '%s' in "
                "region '%s': %s" % (job_name, gcp_project, job_location, e)
            )
            return

        return response