Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
f"the Looker API with the following details: {looker_api_response}\n"
)
sys.exit(error.exit_code)
except SpectaclesException as error:
logger.error(
f"\n{error}\n\n"
+ printer.dim(
"For support, please create an issue at "
"https://github.com/spectacles-ci/spectacles/issues"
)
+ "\n"
)
sys.exit(error.exit_code)
except KeyboardInterrupt as error:
logger.debug(error, exc_info=True)
logger.info("Spectacles was manually interrupted.")
sys.exit(1)
except Exception as error:
logger.debug(error, exc_info=True)
logger.error(
f'\nEncountered unexpected {error.__class__.__name__}: "{error}"\n'
f"Full error traceback logged to file.\n\n"
+ printer.dim(
"For support, please create an issue at "
"https://github.com/spectacles-ci/spectacles/issues"
)
+ "\n"
)
sys.exit(1)
"Check that your credentials are correct and try again."
),
response=response,
)
result = response.json()
if "expires_at" not in result:
# Calculate the expiration time with a one-minute buffer
result["expires_at"] = time.time() + result["expires_in"] - 60
self.access_token = AccessToken(**result)
self.session.headers = { # type: ignore
"Authorization": f"token {self.access_token}"
}
looker_version = self.get_looker_release_version()
logger.info(
f"Connected to Looker version {looker_version} "
f"using Looker API {self.api_version}"
results["errors"],
key=lambda x: (x["model"], x["explore"], x["metadata"]["test_name"]),
)
if errors:
for error in errors:
printer.print_data_test_error(
error["model"],
error["explore"],
error["metadata"]["test_name"],
error["message"],
error["metadata"]["lookml_url"],
)
logger.info("")
raise GenericValidationError
else:
logger.info("")
sql: str,
log_dir: str,
dimension: Optional[str] = None,
lookml_url: Optional[str] = None,
) -> None:
path = model + "/"
if dimension:
path += dimension
else:
path += explore
print_header(red(path), LINE_WIDTH + COLOR_CODE_LENGTH)
wrapped = textwrap.fill(message, LINE_WIDTH)
logger.info(wrapped)
if lookml_url:
logger.info("\n" + f"LookML: {lookml_url}")
file_path = log_sql_error(model, explore, sql, log_dir, dimension)
logger.info("\n" + f"Test SQL: {file_path}")
def print_data_test_error(
model: str, explore: str, test_name: str, message: str, lookml_url: str
) -> None:
path = f"{model}/{explore}/{test_name}"
print_header(red(path), LINE_WIDTH + COLOR_CODE_LENGTH)
wrapped = textwrap.fill(message, LINE_WIDTH)
logger.info(wrapped)
logger.info("\n" + f"LookML: {lookml_url}")
def print_data_test_error(
model: str, explore: str, test_name: str, message: str, lookml_url: str
) -> None:
path = f"{model}/{explore}/{test_name}"
print_header(red(path), LINE_WIDTH + COLOR_CODE_LENGTH)
wrapped = textwrap.fill(message, LINE_WIDTH)
logger.info(wrapped)
logger.info("\n" + f"LookML: {lookml_url}")
log_dir=log_dir,
dimension=error["metadata"].get("dimension"),
lookml_url=error["metadata"].get("lookml_url"),
)
if mode == "batch":
logger.info(
printer.dim(
"\n\nTo determine the exact dimensions responsible for "
f"{'this error' if len(errors) == 1 else 'these errors'}, "
"you can re-run \nSpectacles in single-dimension mode, "
"with `--mode single`.\n\nYou can also run this original "
"validation with `--mode hybrid` to do this automatically."
)
)
logger.info("")
raise GenericValidationError
else:
logger.info("")
)
if mode == "batch":
logger.info(
printer.dim(
"\n\nTo determine the exact dimensions responsible for "
f"{'this error' if len(errors) == 1 else 'these errors'}, "
"you can re-run \nSpectacles in single-dimension mode, "
"with `--mode single`.\n\nYou can also run this original "
"validation with `--mode hybrid` to do this automatically."
)
)
logger.info("")
raise GenericValidationError
else:
logger.info("")