Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.stderr.write(
"Skip processor {}: only one of 'flow_collection' and 'entity' fields "
"allowed".format(p["slug"])
)
continue
p["entity"] = {"type": p.pop("flow_collection")}
if p["type"][-1] != ":":
p["type"] += ":"
if "category" in p and not p["category"].endswith(":"):
p["category"] += ":"
for field in ["input", "output"]:
for schema, _, _ in iterate_schema({}, p[field] if field in p else {}):
if not schema["type"][-1].endswith(":"):
schema["type"] += ":"
# TODO: Check if schemas validate with our JSON meta schema and Processor model docs.
if not self.valid(p, PROCESSOR_SCHEMA):
continue
if "entity" in p:
if "type" not in p["entity"]:
self.stderr.write(
"Skip process {}: 'entity.type' required if 'entity' defined".format(
p["slug"]
)
)
continue
if "input" in p["entity"] and p["entity"].get("always_create", False):
except jsonschema.exceptions.ValidationError as ex:
self.stderr.write(
" VALIDATION ERROR: {}".format(
instance["name"] if "name" in instance else ""
)
)
self.stderr.write(" path: {}".format(ex.path))
self.stderr.write(" message: {}".format(ex.message))
self.stderr.write(" validator: {}".format(ex.validator))
self.stderr.write(" val. value: {}".format(ex.validator_value))
return False
try:
# Check that default values fit field schema.
for field in ["input", "output", "schema"]:
for schema, _, path in iterate_schema({}, instance.get(field, {})):
if "default" in schema:
validate_schema({schema["name"]: schema["default"]}, [schema])
except ValidationError:
self.stderr.write(" VALIDATION ERROR: {}".format(instance["name"]))
self.stderr.write(
" Default value of field '{}' is not valid.".format(path)
)
return False
return True
def validate_range(value, interval, name):
"""Check that given value is inside the specified range."""
if not interval:
return
if value < interval[0] or value > interval[1]:
raise ValidationError(
"Value of field '{}' is out of range. It should be between {} and {}.".format(
name, interval[0], interval[1]
)
)
is_dirty = False
dirty_fields = []
for _schema, _fields, _ in iterate_schema(instance, schema):
name = _schema["name"]
is_required = _schema.get("required", True)
if test_required and is_required and name not in _fields:
is_dirty = True
dirty_fields.append(name)
if name in _fields:
field = _fields[name]
type_ = _schema.get("type", "")
# Treat None as if the field is missing.
if not is_required and field is None:
continue
try:
section = nodes.section(ids=["process-" + slug])
section += nodes.title(name, name)
# Make process header:
section += self.make_process_header(
slug, typ, version, source_uri, description, inputs
)
# Make inputs section:
container_node = nodes.container(classes=["toggle"])
container_header = nodes.paragraph(classes=["header"])
container_header += nodes.strong(text="Input arguments")
container_node += container_header
container_body = nodes.container()
for field_schema, _, path in iterate_schema({}, inputs, ""):
container_body += nodes.strong(text=path)
container_body += self.make_properties_list(field_schema)
container_node += container_body
section += container_node
# Make outputs section:
container_node = nodes.container(classes=["toggle"])
container_header = nodes.paragraph(classes=["header"])
container_header += nodes.strong(text="Output results")
container_node += container_header
container_body = nodes.container()
for field_schema, _, path in iterate_schema({}, outputs, ""):
container_body += nodes.strong(text=path)
container_body += self.make_properties_list(field_schema)
container_body = nodes.container()
for field_schema, _, path in iterate_schema({}, inputs, ""):
container_body += nodes.strong(text=path)
container_body += self.make_properties_list(field_schema)
container_node += container_body
section += container_node
# Make outputs section:
container_node = nodes.container(classes=["toggle"])
container_header = nodes.paragraph(classes=["header"])
container_header += nodes.strong(text="Output results")
container_node += container_header
container_body = nodes.container()
for field_schema, _, path in iterate_schema({}, outputs, ""):
container_body += nodes.strong(text=path)
container_body += self.make_properties_list(field_schema)
container_node += container_body
section += container_node
return [
section,
addnodes.index(entries=[("single", name, "process-" + slug, "", None)]),
]
:param dict inputs: process' inputs
"""
node = addnodes.desc()
signode = addnodes.desc_signature(slug, "")
node.append(signode)
node["objtype"] = node["desctype"] = typ
signode += addnodes.desc_annotation(typ, typ, classes=["process-type"])
signode += addnodes.desc_addname("", "")
signode += addnodes.desc_name(slug + " ", slug + " ")
paramlist = addnodes.desc_parameterlist()
for field_schema, _, _ in iterate_schema({}, inputs, ""):
field_type = field_schema["type"]
field_name = field_schema["name"]
field_default = field_schema.get("default", None)
field_default = "" if field_default is None else "={}".format(field_default)
param = addnodes.desc_parameter("", "", noemph=True)
param += nodes.emphasis(field_type, field_type, classes=["process-type"])
# separate by non-breaking space in the output
param += nodes.strong(text="\xa0\xa0" + field_name)
paramlist += param
signode += paramlist
signode += nodes.reference(
"",