Python voluptuous 模块，Coerce() 实例源码

def get_pagination_options(params, default):
    try:
        opts = voluptuous.Schema({
            voluptuous.Required(
                "limit", default=pecan.request.conf.api.max_limit):
            voluptuous.All(voluptuous.Coerce(int),
                           voluptuous.Range(min=1),
                           voluptuous.Clamp(
                               min=1, max=pecan.request.conf.api.max_limit)),
            "marker": six.text_type,
            voluptuous.Required("sort", default=default):
            voluptuous.All(
                voluptuous.Coerce(arg_to_list),
                [six.text_type]),
        }, extra=voluptuous.REMOVE_EXTRA)(params)
    except voluptuous.Invalid as e:
        abort(400, {"cause": "Argument value error",
                    "reason": str(e)})
    opts['sorts'] = opts['sort']
    del opts['sort']
    return opts

def slugify(value):
    """Coerce a value to a slug."""
    if value is None:
        raise vol.Invalid('Slug should not be None')
    slg = utility_slugify(str(value))
    if len(slg) > 0:
        return slg
    raise vol.Invalid('Unable to slugify {}'.format(value))

def string(value: Any) -> str:
    """Coerce value to string, except for None."""
    if value is not None:
        return str(value)
    raise vol.Invalid('string value is None')

def coerce_datetime(value):
    """Coerce a value to datetime."""
    if isinstance(value, datetime):
        return value

    value = '{}+0000'.format(value)
    try:
        return datetime.strptime(value, DATETIME_FORMAT_MS)
    except (TypeError, ValueError):
        try:
            return datetime.strptime(value, DATETIME_FORMAT)
        except (TypeError, ValueError):
            raise vol.DatetimeInvalid(
                'Value {} does not match expected format {}'.format(
                    value, DATETIME_FORMAT))

def test_can_validate_and_coerce_multiple_provisioner_schemas(self, mock_Provisioner):
        mock_Provisioner.provisioners = {
            'mp1': MockProvisioner1,
            'mp2': MockProvisioner2,
            'mp3': MockProvisioner3}
        schema = get_schema()
        validated = schema({
            'name': 'dummy-test',
            'provisioning': [{
                'type': 'mp1',
                'a': 'dummy',
                'b': '16'
            }, {
                'type': 'mp2',
                'a': 'dummy',
            }, {
                'type': 'mp3',
                'b': 'yes'
            }]
        })
        assert validated == {
            'name': 'dummy-test',
            'provisioning': [{
                'type': 'mp1',
                'a': 'dummy',
                'b': 16  # Check Coerce
            }, {
                'type': 'mp2',
                'a': 'dummy',
            }, {
                'type': 'mp3',
                'b': True  # Check Boolean
            }]
        }

def Size32bit(v):
    """Coerce size to being a 32 bit integer."""
    return as_size(v, max=GiB(4))

def Id(v):
    """Coerce to either a hex UUID, a 2-digit hex value."""
    # Yes, we actually do want this function to raise ValueErrors instead of
    # GadgetSpecificationErrors.
    try:
        return UUID(hex=v)
    except ValueError:
        pass
    mo = re.match('^[a-fA-F0-9]{2}$', v)
    if mo is None:
        raise ValueError(v)
    return mo.group(0).upper()

def post(self):
        enforce("create archive policy", {})
        # NOTE(jd): Initialize this one at run-time because we rely on conf
        conf = pecan.request.conf
        valid_agg_methods = (
            archive_policy.ArchivePolicy.VALID_AGGREGATION_METHODS_VALUES
        )
        ArchivePolicySchema = voluptuous.Schema({
            voluptuous.Required("name"): six.text_type,
            voluptuous.Required("back_window", default=0): PositiveOrNullInt,
            voluptuous.Required(
                "aggregation_methods",
                default=set(conf.archive_policy.default_aggregation_methods)):
            voluptuous.All(list(valid_agg_methods), voluptuous.Coerce(set)),
            voluptuous.Required("definition"):
            voluptuous.All([{
                "granularity": Timespan,
                "points": PositiveNotNullInt,
                "timespan": Timespan,
                }], voluptuous.Length(min=1)),
            })

        body = deserialize_and_validate(ArchivePolicySchema)
        # Validate the data
        try:
            ap = archive_policy.ArchivePolicy.from_dict(body)
        except ValueError as e:
            abort(400, six.text_type(e))
        enforce("create archive policy", ap)
        try:
            ap = pecan.request.indexer.create_archive_policy(ap)
        except indexer.ArchivePolicyAlreadyExists as e:
            abort(409, six.text_type(e))

        location = "/archive_policy/" + ap.name
        set_resp_location_hdr(location)
        pecan.response.status = 201
        return ap

def parse_option(option_string):
    if 'OPTION_NO_VALUE' in option_string:
        option = re.findall(r'\"(.*?)\"', option_string)[0]
        # The options without values seem to still need a value
        # when used with pilight-daemon, but this are not mandatory
        # options
        # E.G.: option 'on' is 'on': 1
        return {vol.Optional(option): vol.Coerce(int)}
    elif 'OPTION_HAS_VALUE' in option_string:
        options = re.findall(r'\"(.*?)\"', option_string)
        option = options[0]
        regex = None
        if len(options) > 1:  # Option has specified value by regex
            regex = options[1]
        if 'JSON_NUMBER' in option_string:
            return {vol.Required(option): vol.Coerce(int)}
        elif 'JSON_STRING' in option_string:
            return {vol.Required(option): vol.Coerce(str)}
        else:
            raise
    elif 'OPTION_OPT_VALUE' in option_string:
        options = re.findall(r'\"(.*?)\"', option_string)
        option = options[0]
        regex = None
        if len(options) > 1:  # Option has specified value by regex
            regex = options[1]
        if 'JSON_NUMBER' in option_string:
            return {vol.Required(option): vol.Coerce(int)}
        elif 'JSON_STRING' in option_string:
            return {vol.Required(option): vol.Coerce(str)}
        else:
            raise
    else:
        print(option_string)
        raise

    raise

def slugify(value):
    """Coerce a value to a slug."""
    if value is None:
        raise vol.Invalid('Slug should not be None')
    slg = util_slugify(str(value))
    if len(slg) > 0:
        return slg
    raise vol.Invalid('Unable to slugify {}'.format(value))

def string(value: Any) -> str:
    """Coerce value to string, except for None."""
    if value is not None:
        return str(value)
    raise vol.Invalid('string value is None')