def run_local_job()

in src/braket/jobs/local/local_job_container.py [0:0]

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

    def run_local_job(self, environment_variables: Dict[str, str]) -> None:
        """Runs a Braket job in a local container.

        Args:
            environment_variables (Dict[str, str]): The environment variables to make available
             as part of running the job.
        """
        start_program_name = self._check_output_formatted(
            ["docker", "exec", self._container_name, "printenv", "SAGEMAKER_PROGRAM"]
        )
        if not start_program_name:
            raise ValueError(
                "Start program not found. "
                "The specified container is not setup to run Braket Jobs. "
                "Please see setup instructions for creating your own containers."
            )

        command = ["docker", "exec", "-w", self.CONTAINER_CODE_PATH]
        command.extend(self._envs_to_list(environment_variables))
        command.append(self._container_name)
        command.append("python")
        command.append(start_program_name)

        try:
            self.run_log = self._check_output_formatted(command)
            print(self.run_log)
        except subprocess.CalledProcessError as e:
            self.run_log = e.output.decode("utf-8").strip()
            self._logger.error(self.run_log)