Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _validate_docs_schema(self, doc, schema, name, error_code):
# TODO: Add line/col
errors = []
try:
schema(doc)
except Exception as e:
for error in e.errors:
error.data = doc
errors.extend(e.errors)
for error in errors:
path = [str(p) for p in error.path]
if isinstance(error.data, dict):
error_message = humanize_error(error.data, error)
else:
error_message = error
if path:
combined_path = '%s.%s' % (name, '.'.join(path))
else:
combined_path = name
self.reporter.error(
path=self.object_path,
code=error_code,
msg='%s: %s' % (combined_path, error_message)
)
This method must be run in the event loop.
"""
message = f"Invalid config for [{domain}]: "
if isinstance(ex, vol.Invalid):
if "extra keys not allowed" in ex.error_message:
message += (
"[{option}] is an invalid option for [{domain}]. "
"Check: {domain}->{path}.".format(
option=ex.path[-1],
domain=domain,
path="->".join(str(m) for m in ex.path),
)
)
else:
message += "{}.".format(humanize_error(config, ex))
else:
message += str(ex)
try:
domain_config = config.get(domain, config)
except AttributeError:
domain_config = config
message += " (See {}, line {}). ".format(
getattr(domain_config, "__config_file__", "?"),
getattr(domain_config, "__line__", "?"),
)
if domain != CONF_CORE:
integration = domain.split(".")[-1]
message += (
def save_data(self) -> None:
"""Store data to configuration file."""
# Validate
try:
self._data = self._schema(self._data)
except vol.Invalid as ex:
_LOGGER.error("Can't parse data: %s", humanize_error(self._data, ex))
# Load last valid data
_LOGGER.warning("Reset %s to last version", self._file)
self.read_data()
else:
# write
try:
write_json_file(self._file, self._data)
except JsonFileError:
pass
async def api_validate(schema, request):
"""Validate request data with schema."""
data = await request.json(loads=json_loads)
try:
data = schema(data)
except vol.Invalid as ex:
raise RuntimeError(humanize_error(data, ex)) from None
return data
def reset_data(self) -> None:
"""Reset JSON file to default."""
try:
self._data = self._schema({})
except vol.Invalid as ex:
_LOGGER.error(
"Can't reset %s: %s", self._file, humanize_error(self._data, ex)
)
async def validate_and_run(hass, config_entry, data):
"""Validate input and call handler."""
try:
data = schema(data)
except vol.Invalid as ex:
err = vol.humanize.humanize_error(data, ex)
_LOGGER.error("Received invalid webhook payload: %s", err)
return empty_okay_response()
return await func(hass, config_entry, data)
def read_data(self) -> None:
"""Read JSON file & validate."""
if self._file.is_file():
try:
self._data = read_json_file(self._file)
except JsonFileError:
self._data = {}
# Validate
try:
self._data = self._schema(self._data)
except vol.Invalid as ex:
_LOGGER.error(
"Can't parse %s: %s", self._file, humanize_error(self._data, ex)
)
# Reset data to default
_LOGGER.warning("Reset %s to default", self._file)
self._data = self._schema({})
def _read_addons_folder(self, path, repository):
"""Read data from add-ons folder."""
for addon in path.glob("**/config.json"):
try:
addon_config = read_json_file(addon)
except JsonFileError:
_LOGGER.warning("Can't read %s from repository %s", addon, repository)
continue
# validate
try:
addon_config = SCHEMA_ADDON_CONFIG(addon_config)
except vol.Invalid as ex:
_LOGGER.warning(
"Can't read %s: %s", addon, humanize_error(addon_config, ex)
)
continue
# Generate slug
addon_slug = "{}_{}".format(repository, addon_config[ATTR_SLUG])
# store
addon_config[ATTR_REPOSITORY] = repository
addon_config[ATTR_LOCATON] = str(addon.parent)
self.addons[addon_slug] = addon_config