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_invalid(self):
with self.assertRaises(ValidationError):
GitPackageContract.from_dict(
{'git': 'http://example.com', 'version': '0.0.1'}
)
def assert_fails_validation(self, dct, cls=None):
if cls is None:
cls = self.ContractType
with self.assertRaises(ValidationError):
cls.from_dict(dct)
}
format_stats will convert the dict into a StatsDict with keys of 'encoded'
and 'size'.
"""
stats_collector: StatsDict = {}
base_keys = {k.split(':')[0] for k in stats}
for key in base_keys:
dct: PrimitiveDict = {'id': key}
for subkey in ('label', 'value', 'description', 'include'):
dct[subkey] = stats['{}:{}'.format(key, subkey)]
try:
stats_item = StatsItem.from_dict(dct)
except ValidationError:
continue
if stats_item.include:
stats_collector[key] = stats_item
# we always have a 'has_stats' field, it's never included
has_stats = StatsItem(
id='has_stats',
label='Has Stats?',
value=len(stats_collector) > 0,
description='Indicates whether there are statistics for this table',
include=False,
)
stats_collector['has_stats'] = has_stats
return stats_collector
def read_yaml_sources(
self, yaml: YamlBlock
) -> Iterable[SourceTarget]:
path = yaml.path.original_file_path
yaml_key = 'sources'
for data in self._get_dicts_for(yaml, yaml_key):
try:
data = self._renderer.render_schema_source(data)
source = UnparsedSourceDefinition.from_dict(data)
except (ValidationError, JSONValidationException) as exc:
msg = error_context(path, yaml_key, data, exc)
raise CompilationException(msg) from exc
else:
for table in source.tables:
yield SourceTarget(source, table)
def __call__(self, **kwargs: Dict[str, Any]) -> JsonSchemaMixin:
try:
params = self.get_parameters().from_dict(kwargs)
except ValidationError as exc:
raise TypeError(exc) from exc
self.set_args(params)
return self.handle_request()
def validator_error_message(exc):
"""Given a hologram.ValidationError (which is basically a
jsonschema.ValidationError), return the relevant parts as a string
"""
if not isinstance(exc, hologram.ValidationError):
return str(exc)
path = "[%s]" % "][".join(map(repr, exc.relative_path))
return 'at path {}: {}'.format(path, exc.message)
def to_python(self, value) -> NoValue:
if (
not isinstance(value, dict) or
'novalue' not in value or
value['novalue'] != 'novalue'
):
raise ValidationError('Got invalid NoValue: {}'.format(value))
return NoValue()
'database': self.default_database,
'fqn': config.fqn,
'name': name,
'root_path': self.project.project_root,
'resource_type': self.resource_type,
'path': path,
'original_file_path': block.path.original_file_path,
'package_name': self.project.project_name,
'raw_sql': block.contents,
'unique_id': self.generate_unique_id(name),
'config': self.config_dict(config),
}
dct.update(kwargs)
try:
return self.parse_from_dict(dct)
except ValidationError as exc:
msg = validator_error_message(exc)
# this is a bit silly, but build an UnparsedNode just for error
# message reasons
node = self._create_error_node(
name=block.name,
path=path,
original_file_path=block.path.original_file_path,
raw_sql=block.contents,
)
raise CompilationException(msg, node=node)
def error_context(
path: str,
key: str,
data: Any,
cause: Union[str, ValidationException, JSONValidationException]
) -> str:
"""Provide contextual information about an error while parsing
"""
if isinstance(cause, str):
reason = cause
elif isinstance(cause, ValidationError):
reason = validator_error_message(cause)
else:
reason = cause.msg
return (
'Invalid {key} config given in {path} @ {key}: {data} - {reason}'
.format(key=key, path=path, data=data, reason=reason)
)
def _collect_parameters(self):
# both get_parameters and the argparse can raise a TypeError.
cls: Type[RPCParameters] = self.task.get_parameters()
try:
return cls.from_dict(self.task_kwargs)
except ValidationError as exc:
# raise a TypeError to indicate invalid parameters so we get a nice
# error from our json-rpc library
raise TypeError(exc) from exc