Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_git(self):
try:
dbt.clients.system.run_cmd(os.getcwd(), ['git', '--help'])
except dbt.exceptions.ExecutableError as exc:
self.messages.append('Error from git --help: {!s}'.format(exc))
return red('ERROR')
return green('OK found')
def test_git(self):
try:
dbt.clients.system.run_cmd(os.getcwd(), ['git', '--help'])
except dbt.exceptions.ExecutableError as exc:
self.messages.append('Error from git --help: {!s}'.format(exc))
return red('ERROR')
return green('OK found')
def _handle_internal_exception(self, e, ctx):
build_path = self.node.build_path
prefix = 'Internal error executing {}'.format(build_path)
error = "{prefix}\n{error}\n\n{note}".format(
prefix=dbt.ui.printer.red(prefix),
error=str(e).strip(),
note=INTERNAL_ERROR_STRING
)
logger.debug(error, exc_info=True)
return str(e)
def _load_project(self):
if not os.path.exists(self.project_path):
self.project_fail_details = FILE_NOT_FOUND
return red('ERROR not found')
try:
self.project = Project.from_current_directory(self.cli_vars)
except dbt.exceptions.DbtConfigError as exc:
self.project_fail_details = str(exc)
return red('ERROR invalid')
return green('OK found and valid')
def _load_profile(self):
if not os.path.exists(self.profile_path):
self.profile_fail_details = FILE_NOT_FOUND
self.messages.append(MISSING_PROFILE_MESSAGE.format(
path=self.profile_path, url=ProfileConfigDocs
))
return red('ERROR not found')
try:
raw_profile_data = load_yaml_text(
dbt.clients.system.load_file_contents(self.profile_path)
)
except Exception:
pass # we'll report this when we try to load the profile for real
else:
if isinstance(raw_profile_data, dict):
self.raw_profile_data = raw_profile_data
self.profile_name = self._choose_profile_name()
self.target_name = self._choose_target_name()
try:
self.profile = QueryCommentedProfile.from_args(
self.args, self.profile_name
def _target_found(self):
requirements = (self.raw_profile_data and self.profile_name and
self.target_name)
if not requirements:
return red('ERROR not found')
if self.profile_name not in self.raw_profile_data:
return red('ERROR not found')
profiles = self.raw_profile_data[self.profile_name]['outputs']
if self.target_name not in profiles:
return red('ERROR not found')
return green('OK found')
def _load_project(self):
if not os.path.exists(self.project_path):
self.project_fail_details = FILE_NOT_FOUND
return red('ERROR not found')
try:
self.project = Project.from_current_directory(self.cli_vars)
except dbt.exceptions.DbtConfigError as exc:
self.project_fail_details = str(exc)
return red('ERROR invalid')
return green('OK found and valid')
def _target_found(self):
requirements = (self.raw_profile_data and self.profile_name and
self.target_name)
if not requirements:
return red('ERROR not found')
if self.profile_name not in self.raw_profile_data:
return red('ERROR not found')
profiles = self.raw_profile_data[self.profile_name]['outputs']
if self.target_name not in profiles:
return red('ERROR not found')
return green('OK found')
def _handle_internal_exception(self, e, ctx):
build_path = self.node.build_path
prefix = 'Internal error executing {}'.format(build_path)
error = "{prefix}\n{error}\n\n{note}".format(
prefix=dbt.ui.printer.red(prefix),
error=str(e).strip(),
note=INTERNAL_ERROR_STRING)
logger.debug(error)
return dbt.compat.to_string(e)
def _profile_found(self):
if not self.raw_profile_data:
return red('ERROR not found')
if self.profile_name in self.raw_profile_data:
return green('OK found')
else:
return red('ERROR not found')