How to use the schema.And function in schema

To help you get started, we’ve selected a few schema examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github alteryx / promote-python / tests / tests.py View on Github external
        @promote.validate_json(Schema({'name': And(str, len)}))
        def test_function(data):
            return data
github apls777 / spotty / spotty / config / validation.py View on Github external
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('/')),
                                                          ),
github Nachtfeuer / pipeline / spline / components / config.py View on Github external
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
        })
github carsdotcom / skelebot / skelebot / components / jupyter.py View on Github external
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
        """
github pioneers / PieCentral / runtime / runtime / messaging / device.py View on Github external
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'):
github apls777 / spotty / spotty / config / validation.py View on Github external
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 '
github cisagov / travis-wait-improved / src / example / example.py View on Github external
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
        }
    )
github supertylerc / network-static-analysis / nsa / api.py View on Github external
"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),
    }
github Nachtfeuer / pipeline / spline / components / config.py View on Github external
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
        })
github botfront / rasa-addons / rasa_addons / disambiguation.py View on Github external
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)