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
python类Coerce()的实例源码
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')
