Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@promote.validate_json(Schema({'name': And(str, len)}))
def test_function(data):
return data
default='create_snapshot'): And(str, lambda x: x in ['create_snapshot', 'update_snapshot',
'retain', 'delete'],
error='Incorrect value for "deletionPolicy".'
),
}],
And(lambda x: len(x) < 12, error='Maximum 11 volumes are supported at the moment.'),
),
'docker': And(
{
Optional('image', default=''): str,
Optional('file', default=''): And(str, # TODO: a proper regex that the filename is valid
Regex(r'^[\w\.\/@-]*$',
error='Invalid name for a Dockerfile'),
And(lambda x: not x.endswith('/'),
error='Invalid name for a Dockerfile'),
And(lambda x: not os.path.isabs(x),
error='Path to the Dockerfile should be relative to the '
'project\'s root directory.'),
And(lambda x: os.path.isfile(os.path.join(project_dir, x)),
error='Dockerfile not found.'),
),
Optional('workingDir', default=''): And(str,
And(os.path.isabs,
error='Use an absolute path when specifying a '
'working directory'),
),
Optional('dataRoot', default=''): And(str,
And(os.path.isabs,
error='Use an absolute path when specifying a Docker '
'data root directory'),
Use(lambda x: x.rstrip('/')),
),
def schema():
"""Provide schema for shell configuration."""
return Schema({
'script': And(Or(type(' '), type(u' ')), len),
Optional('title', default=''): str,
Optional('model', default={}): {Optional(And(str, len)): object},
Optional('env', default={}): {Optional(And(str, len)): And(str, len)},
Optional('item', default=None): object,
Optional('dry_run', default=False): bool,
Optional('debug', default=False): bool,
Optional('strict', default=False): bool,
Optional('variables', default={}): {
Optional(And(Or(type(' '), type(u' ')), len, Regex(r'([a-zA-Z][_a-zA-Z]*)'))):
Or(type(' '), type(u' '))
},
Optional('temporary_scripts_path', default=''): Or(type(''), type(u'')),
Optional('internal', default=False): bool
})
COMMAND_TEMPLATE = "jupyter notebook --ip=0.0.0.0 --port=8888 --allow-root --notebook-dir={folder}"
class Jupyter(Component):
"""
Jupyter Class
Provides the ability to spin up Jupyter notebooks inside a Docker container with the exact
configuration as the rest of the project and it's jobs
"""
activation = Activation.PROJECT
commands = ["jupyter"]
schema = Schema({
Optional('port'): And(int, error='Jupyter \'port\' must be an Integer'),
Optional('folder'): And(str, error='Jupyter \'folder\' must be a String'),
}, ignore_extra_keys=True)
port = None
folder = None
def __init__(self, port=8888, folder="."):
"""Initialize the class with simple default values for port and folder"""
self.port = port
self.folder = folder
def addParsers(self, subparsers):
"""
SkeleParser Hook
Adds a parser for the jupter command the starts up the notebooks in Docker
"""
raise RuntimeBaseException('Maxmimum number of Smart Sensor parameters exceeded')
extra_fields = [
('write', ctypes.c_uint16),
('read', ctypes.c_uint16),
('send', ctypes.c_uint16),
('delay', ctypes.c_uint16),
('subscription', ctypes.c_uint16),
('uid', packetlib.SmartSensorUID),
]
return DeviceStructure.make_type(name, type_id, params, *extra_fields, base_cls=cls)
DEVICE_SCHEMA = Schema({
And(Use(str), VALID_NAME): { # Protocol name
And(Use(str), VALID_NAME): { # Device name
'id': And(Use(int), lambda type_id: 0 <= type_id < 0xFFFF),
'params': [{
'name': Use(str),
'type': And(Use(str), Use(lambda dev_type: getattr(ctypes, f'c_{dev_type}'))),
Optional('lower'): Use(float),
Optional('upper'): Use(float),
Optional('readable'): Use(bool),
Optional('writeable'): Use(bool),
Optional('subscribed'): Use(bool),
}]
}
}
})
DEVICES = {}
def load_device_types(schema: dict, sensor_protocol: str = 'smartsensor'):
def validate_old_config(data, project_dir):
schema = Schema({
'project': {
'name': And(str, Regex(r'^[a-zA-Z0-9][a-zA-Z0-9-]{,26}[a-zA-Z0-9]$')),
'remoteDir': And(str,
And(os.path.isabs,
error='Use an absolute path when specifying a remote directory'),
Use(lambda x: x.rstrip('/'))
),
Optional('syncFilters', default=[]): [And(
{
Optional('exclude'): [And(str, len)],
Optional('include'): [And(str, len)],
},
And(lambda x: x, error='Either "exclude" or "include" filter should be specified.'),
And(lambda x: not ('exclude' in x and 'include' in x), error='"exclude" and "include" filters should '
'be specified as different list items.'),
)]
},
'instance': {
'region': And(str, len),
Optional('availabilityZone', default=''): str,
Optional('subnetId', default=''): str,
'instanceType': str,
Optional('onDemandInstance', default=False): bool,
Optional('amiName', default='SpottyAMI'): str,
Optional('keyName', default=''): str,
Optional('rootVolumeSize', default=0): And(Or(int, str), Use(str),
Regex(r'^\d+$', error='Incorrect value for "rootVolumeSize".'),
Use(int),
And(lambda x: x > 0,
error='"rootVolumeSize" should be greater than 0 or should '
def main() -> int:
"""Set up logging and call the example function."""
args: Dict[str, str] = docopt.docopt(__doc__, version=__version__)
# Validate and convert arguments as needed
schema: Schema = Schema(
{
"--log-level": And(
str,
Use(str.lower),
lambda n: n in ("debug", "info", "warning", "error", "critical"),
error="Possible values for --log-level are "
+ "debug, info, warning, error, and critical.",
),
"": Use(int, error=" must be an integer."),
"": And(
Use(int),
lambda n: n != 0,
error=" must be an integer that is not 0.",
),
str: object, # Don't care about other keys, if any
}
)
"match": And(str, len),
"next_lines": And(
list, lambda l: all(isinstance(i, str) for i in l)
),
"wiki": str,
Optional("cve"): And(str, Use(str.upper)),
Optional("score"): And(Or(int, float), lambda n: n > 0),
}
]
)
INPUT_SCHEMA = Schema(
{
"device": And(str, len),
"platform": And(str, len),
"contents": And(list, len),
}
def schema():
"""Provide schema for shell configuration."""
return Schema({
'script': And(Or(type(' '), type(u' ')), len),
Optional('title', default=''): str,
Optional('model', default={}): {Optional(And(str, len)): object},
Optional('env', default={}): {Optional(And(str, len)): And(str, len)},
Optional('item', default=None): object,
Optional('dry_run', default=False): bool,
Optional('debug', default=False): bool,
Optional('strict', default=False): bool,
Optional('variables', default={}): {
Optional(And(Or(type(' '), type(u' ')), len, Regex(r'([a-zA-Z][_a-zA-Z]*)'))):
Or(type(' '), type(u' '))
},
Optional('temporary_scripts_path', default=''): Or(type(''), type(u'')),
Optional('internal', default=False): bool
})
def __init__(self, disamb_rule=None, fallback_rule=None):
self.disamb_schema = Schema({
"trigger": And(str, len),
Optional("max_suggestions", default=2): int,
Optional("slot_name"): str,
"display": {
Optional("intro_template"): And(str, len),
"text_template": And(str, len, error="text_template is required"),
Optional("button_title_template_prefix", default="utter_disamb"): And(str, len),
Optional("fallback_button"): {
"title": And(str, len, error="fallback button title is required"),
"payload": And(str, len, error="fallback button payload is required")
},
Optional("exclude", default=[]): list
}
})
self.fallback_schema = Schema({
"trigger": And(str, len),
Optional("slot_name"): str,
"display": {
"text": And(str, len),
Optional("buttons"):
[{"title": And(str, len, error="button title is required"),
"payload": And(str, len, error="button title is required")}]
}
})
if disamb_rule:
self.disamb_schema.validate(disamb_rule)