Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_fallback_option_works(self):
parser = config.Fallback(config.Percent, config.Float)
self.assertAlmostEqual(parser(".44"), 0.44)
# quiet kazoo's verbose logs a bit
logging.getLogger("kazoo").setLevel(logging.WARNING)
parser = configparser.RawConfigParser()
parser.read_file(args.config_file)
watcher_config = dict(parser.items("live-data"))
cfg = config.parse_config(
watcher_config,
{
"nodes": config.DictOf(
{
"source": config.String,
"dest": config.String,
"owner": config.Optional(config.UnixUser),
"group": config.Optional(config.UnixGroup),
"mode": config.Optional(config.Integer(base=8), default=0o400), # type: ignore
}
)
},
)
# pylint: disable=maybe-no-member
nodes = cfg.nodes.values()
secrets = secrets_store_from_config(watcher_config, timeout=30)
zookeeper = zookeeper_client_from_config(secrets, watcher_config, read_only=True)
zookeeper.start()
try:
watch_zookeeper_nodes(zookeeper, nodes)
finally:
zookeeper.stop()
"socket_timeout": config.Optional(config.Timespan, default=None),
}
)
options = parser.parse(prefix[:-1], app_config)
if options.max_connections is not None:
kwargs.setdefault("max_connections", options.max_connections)
if options.socket_connect_timeout is not None:
kwargs.setdefault("socket_connect_timeout", options.socket_connect_timeout.total_seconds())
if options.socket_timeout is not None:
kwargs.setdefault("socket_timeout", options.socket_timeout.total_seconds())
return redis.BlockingConnectionPool.from_url(options.url, **kwargs)
class RedisClient(config.Parser):
"""Configure a Redis client.
This is meant to be used with
:py:meth:`baseplate.Baseplate.configure_context`.
See :py:func:`pool_from_config` for available configuration settings.
"""
def __init__(self, **kwargs: Any):
self.kwargs = kwargs
def parse(self, key_path: str, raw_config: config.RawConfig) -> "RedisContextFactory":
connection_pool = pool_from_config(raw_config, f"{key_path}.", **self.kwargs)
return RedisContextFactory(connection_pool)
error_reporter_from_config(app_config, __name__)
:param raw_config: The application configuration which should have
settings for the error reporter.
:param module_name: ``__name__`` of the root module of the application.
"""
cfg = config.parse_config(
raw_config,
{
"sentry": {
"dsn": config.Optional(config.String, default=None),
"site": config.Optional(config.String, default=None),
"environment": config.Optional(config.String, default=None),
"include_paths": config.Optional(config.String, default=None),
"exclude_paths": config.Optional(config.String, default=None),
"ignore_exceptions": config.Optional(config.TupleOf(config.String), default=[]),
"sample_rate": config.Optional(config.Percent, default=1),
"processors": config.Optional(
config.TupleOf(config.String),
default=["raven.processors.SanitizePasswordsProcessor"],
),
}
},
)
application_module = sys.modules[module_name]
module_path = os.path.abspath(application_module.__file__)
directory = os.path.dirname(module_path)
release = None
while directory != "/":