Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
log_level = vol.In(["debug", "info", "warning", "error", "critical"])
def dns_url(url: str) -> str:
""" takes a DNS url (str) and validates that it matches the scheme dns://."""
if not url.lower().startswith("dns://"):
raise vol.Invalid("Doesn't start with dns://")
address: str = url[6:] # strip the dns:// off
try:
ipaddress.ip_address(address) # matches ipv4 or ipv6 addresses
except ValueError:
raise vol.Invalid("Invalid DNS URL: {}".format(url))
return url
dns_server_list = vol.All(vol.Length(max=8), [dns_url])
def validate_repository(repository: str) -> str:
"""Validate a valid repository."""
data = RE_REPOSITORY.match(repository)
if not data:
raise vol.Invalid("No valid repository format!")
# Validate URL
# pylint: disable=no-value-for-parameter
vol.Url()(data.group("url"))
return repository
# pylint: disable=no-value-for-parameter
Required('bind'): str,
Required('port'): int,
'append_path': [str],
'environ': [{
Required('key'): str,
Required('value'): str
}],
'static': [{
Required('url'): str,
Required('path'): str
}],
Required('default_app'): str,
Required('sub_modules'): All([{
Required('url'): str,
Required('module'): str,
}], Length(min=1)),
}
def __init__(self, config_path):
try:
self.base_path = os.path.dirname(config_path)
parsed_config = yaml.load(file(config_path))
except IOError, exp:
raise prestans.devel.exception.Base("[error] unable to read configuration at %s" % config_path)
try:
schema = Schema(Configuration._SCHEMA)
validated_config = schema(parsed_config)
except MultipleInvalid, exp:
raise prestans.devel.exception.Base("[error/config] %s" % str(exp), 2)
),
vol.Optional(CONF_DRIVER): str,
vol.Optional(CONF_BUCKLING): bool,
vol.Optional(CONF_REACTION_DRIVER): str,
vol.Optional(CONF_BLOCKTYPES): [str],
vol.Optional(CONF_HOMOGBLOCK): bool,
vol.Optional(CONF_EXTERNAL_DRIVER, default=True): bool,
vol.Optional(CONF_INTERNAL_RINGS): vol.Coerce(int),
vol.Optional(CONF_EXTERNAL_RINGS): vol.Coerce(int),
vol.Optional(CONF_MERGE_INTO_CLAD): [str],
vol.Optional(CONF_FILE_LOCATION): [str],
vol.Optional(CONF_MESH_PER_CM): vol.Coerce(float),
}
)
XS_SCHEMA = vol.Schema({vol.All(str, vol.Length(min=1, max=2)): SINGLE_XS_SCHEMA})
class XSModelingOptions:
"""Advanced cross section modeling options for a particular XS ID."""
def __init__(
self,
xsID,
geometry=None,
blockRepresentation=None,
driverID=None,
criticalBuckling=None,
nuclideReactionDriver=None,
validBlockTypes=None,
externalDriver=True,
useHomogenizedBlockComposition=False,
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_TOKEN
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.util import color, dt
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "Xiaomi Philips Light"
DATA_KEY = "light.xiaomi_miio"
CONF_MODEL = "model"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_TOKEN): vol.All(cv.string, vol.Length(min=32, max=32)),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MODEL): vol.In(
[
"philips.light.sread1",
"philips.light.ceiling",
"philips.light.zyceiling",
"philips.light.moonlight",
"philips.light.bulb",
"philips.light.candle",
"philips.light.candle2",
"philips.light.mono1",
"philips.light.downlight",
]
),
}
)
})
schema_provides = Schema({
Any('mimetypes',
'binaries',
'libraries',
'python3',
'python2',
'firmware',
'modaliases',
'fonts'): All(list, [str], Length(min=1)),
'dbus': All(list, Length(min=1), [schema_provides_dbus]),
})
schema_keywords = Schema({
Required('C'): All(list, [str], Length(min=1), msg="Must have an unlocalized 'C' key"),
dict: All(list, [str], Length(min=1)),
}, extra = True)
schema_translated = Schema({
Required('C'): All(str, Length(min=1), msg="Must have an unlocalized 'C' key"),
dict: All(str, Length(min=1)),
}, extra = True)
schema_image = Schema({
Required('width'): All(int, Range(min=10)),
Required('height'): All(int, Range(min=10)),
Required('url'): All(str, Url()),
})
schema_screenshots = Schema({
Required('default', default=False): All(bool),
def MetricSchema(definition):
creator = pecan.request.auth_helper.get_current_user(
pecan.request)
# First basic validation
schema = voluptuous.Schema({
"archive_policy_name": six.text_type,
"resource_id": functools.partial(ResourceID, creator=creator),
"name": six.text_type,
voluptuous.Optional("unit"):
voluptuous.All(six.text_type, voluptuous.Length(max=31)),
})
definition = schema(definition)
archive_policy_name = definition.get('archive_policy_name')
name = definition.get('name')
if name and '/' in name:
abort(400, "'/' is not supported in metric name")
if archive_policy_name is None:
try:
ap = pecan.request.indexer.get_archive_policy_for_metric(name)
except indexer.NoArchivePolicyRuleMatch:
# NOTE(jd) Since this is a schema-like function, we
# should/could raise ValueError, but if we do so, voluptuous
# just returns a "invalid value" with no useful message – so we
# prefer to use abort() to make sure the user has the right
# error message
raise voluptuous.Invalid("Expected a tuple/list, got a %s" % type(v))
elif not v:
raise voluptuous.Invalid("Operation must not be empty")
elif len(v) < 2:
raise voluptuous.Invalid("Operation need at least one argument")
elif v[0] != u"metric":
# NOTE(sileht): this error message doesn't looks related to "metric",
# but because that the last schema validated by voluptuous, we have
# good chance (voluptuous.Any is not predictable) to print this
# message even if it's an other operation that invalid.
raise voluptuous.Invalid("'%s' operation invalid" % v[0])
return [u"metric"] + voluptuous.Schema(voluptuous.Any(
voluptuous.ExactSequence([six.text_type, six.text_type]),
voluptuous.All(
voluptuous.Length(min=1),
[voluptuous.ExactSequence([six.text_type, six.text_type])],
)), required=True)(v[1:])
})
ops = [post_op]
# if comment_options are used, add a new op to the transaction
if comment_options or beneficiaries:
options = keep_in_dict(comment_options or {}, [
'max_accepted_payout', 'percent_steem_dollars', 'allow_votes',
'allow_curation_rewards', 'extensions'
])
# override beneficiaries extension
if beneficiaries:
# validate schema
# or just simply vo.Schema([{'account': str, 'weight': int}])
schema = vo.Schema([{
vo.Required('account'):
vo.All(str, vo.Length(max=16)),
vo.Required('weight', default=10000):
vo.All(int, vo.Range(min=1, max=10000))
}])
schema(beneficiaries)
options['beneficiaries'] = beneficiaries
default_max_payout = "1000000.000 SBD"
comment_op = operations.CommentOptions(
**{
"author":
author,
"permlink":
permlink,
"max_accepted_payout":
options.get("max_accepted_payout", default_max_payout),
"percent_steem_dollars":
voluptuous.Length(min=0, max=1),
{
voluptuous.Any(
u"=", u"==", u"eq",
u"<", u"lt",
u">", u"gt",
u"<=", u"≤", u"le",
u">=", u"≥", u"ge",
u"!=", u"≠", u"ne",
): voluptuous.All(
voluptuous.Length(min=1, max=1),
{"id": _ResourceUUID,
NotIDKey: ResourceSearchSchemaAttributeValue},
),
u"like": voluptuous.All(
voluptuous.Length(min=1, max=1),
{NotIDKey: ResourceSearchSchemaAttributeValue},
),
u"in": voluptuous.All(
voluptuous.Length(min=1, max=1),
{"id": voluptuous.All(
[_ResourceUUID],
voluptuous.Length(min=1)),
NotIDKey: voluptuous.All(
[ResourceSearchSchemaAttributeValue],
voluptuous.Length(min=1))}
),
voluptuous.Any(
u"and", u"∨",
u"or", u"∧",
): voluptuous.All(
[ResourceSearchSchema], voluptuous.Length(min=1)
CONF_TURN_ON_COMMAND = 'turn_on_command'
CONF_TURN_ON_PARAMETERS = 'turn_on_parameters'
CONF_TURN_OFF_COMMAND = 'turn_off_command'
CONF_TURN_OFF_PARAMETERS = 'turn_off_parameters'
CONF_STATE_PROPERTY = 'state_property'
CONF_STATE_ON_VALUE = 'state_on_value'
CONF_STATE_DELAY_VALUE = 'state_delay_value'
CONF_STATE_OFF_VALUE = 'state_off_value'
CONF_UPDATE_INSTANT = 'update_instant'
ATTR_STATE_PROPERTY = CONF_STATE_PROPERTY
ATTR_STATE_VALUE = 'state_value'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_TOKEN): vol.All(cv.string, vol.Length(min=32, max=32)),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_TURN_ON_COMMAND, default='set_arming'): cv.string,
vol.Optional(CONF_TURN_ON_PARAMETERS, default=['on']):
vol.All(cv.ensure_list, [cv.string]),
vol.Optional(CONF_TURN_OFF_COMMAND, default='set_arming'): cv.string,
vol.Optional(CONF_TURN_OFF_PARAMETERS, default=['off']):
vol.All(cv.ensure_list, [cv.string]),
vol.Optional(CONF_STATE_PROPERTY, default=''): cv.string,
vol.Optional(CONF_STATE_ON_VALUE, default='on'): cv.string,
vol.Optional(CONF_STATE_DELAY_VALUE, default='oning'): cv.string,
vol.Optional(CONF_STATE_OFF_VALUE, default='off'): cv.string,
vol.Optional(CONF_UPDATE_INSTANT, default=True): cv.boolean,
})
REQUIREMENTS = ['python-miio>=0.4.5']