本文整理汇总了Python中redash.utils.configuration.ConfigurationContainer类的典型用法代码示例。如果您正苦于以下问题：Python ConfigurationContainer类的具体用法？Python ConfigurationContainer怎么用？Python ConfigurationContainer使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了ConfigurationContainer类的16个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: post
def post(self):
req = request.get_json(True)
require_fields(req, ('options', 'name', 'type'))
schema = get_configuration_schema_for_query_runner_type(req['type'])
if schema is None:
abort(400)
config = ConfigurationContainer(filter_none(req['options']), schema)
# from IPython import embed
# embed()
if not config.is_valid():
abort(400)
try:
datasource = models.DataSource.create_with_group(org=self.current_org,
name=req['name'],
type=req['type'],
options=config)
models.db.session.commit()
except IntegrityError as e:
if req['name'] in e.message:
abort(400, message="Data source with the name {} already exists.".format(req['name']))
abort(400)
self.record_event({
'action': 'create',
'object_id': datasource.id,
'object_type': 'datasource'
})
return datasource.to_dict(all=True)
开发者ID:getredash，项目名称:redash，代码行数:34，代码来源:data_sources.py
示例2: post
def post(self):
req = request.get_json(True)
require_fields(req, ('options', 'name', 'type'))
schema = get_configuration_schema_for_destination_type(req['type'])
if schema is None:
abort(400)
config = ConfigurationContainer(req['options'], schema)
if not config.is_valid():
abort(400)
destination = models.NotificationDestination(org=self.current_org,
name=req['name'],
type=req['type'],
options=config,
user=self.current_user)
try:
models.db.session.add(destination)
models.db.session.commit()
except IntegrityError as e:
if 'name' in e.message:
abort(400, message=u"Alert Destination with the name {} already exists.".format(req['name']))
abort(500)
return destination.to_dict(all=True)
开发者ID:ariarijp，项目名称:redash，代码行数:27，代码来源:destinations.py
示例3: post
def post(self):
req = request.get_json(True)
required_fields = ('options', 'name', 'type')
for f in required_fields:
if f not in req:
abort(400)
schema = get_configuration_schema_for_type(req['type'])
if schema is None:
abort(400)
config = ConfigurationContainer(req['options'], schema)
if not config.is_valid():
abort(400)
datasource = models.DataSource.create_with_group(org=self.current_org,
name=req['name'],
type=req['type'],
options=config)
self.record_event({
'action': 'create',
'object_id': datasource.id,
'object_type': 'datasource'
})
return datasource.to_dict(all=True)
开发者ID:Tomer1510，项目名称:redash，代码行数:26，代码来源:data_sources.py
示例4: new
def new(name=None, type=None, options=None):
"""Create new data source."""
if name is None:
name = click.prompt("Name")
if type is None:
print "Select type:"
for i, query_runner_name in enumerate(query_runners.keys()):
print "{}. {}".format(i + 1, query_runner_name)
idx = 0
while idx < 1 or idx > len(query_runners.keys()):
idx = click.prompt("[{}-{}]".format(1, len(query_runners.keys())), type=int)
type = query_runners.keys()[idx - 1]
else:
validate_data_source_type(type)
if options is None:
query_runner = query_runners[type]
schema = query_runner.configuration_schema()
types = {
'string': unicode,
'number': int,
'boolean': bool
}
options_obj = {}
for k, prop in schema['properties'].iteritems():
required = k in schema.get('required', [])
default_value = "<<DEFAULT_VALUE>>"
if required:
default_value = None
prompt = prop.get('title', k.capitalize())
if required:
prompt = "{} (required)".format(prompt)
else:
prompt = "{} (optional)".format(prompt)
value = click.prompt(prompt, default=default_value, type=types[prop['type']], show_default=False)
if value != default_value:
options_obj[k] = value
options = ConfigurationContainer(options_obj, schema)
if not options.is_valid():
print "Error: invalid configuration."
exit()
print "Creating {} data source ({}) with options:\n{}".format(type, name, options)
data_source = models.DataSource.create(name=name,
type=type,
options=options,
org=models.Organization.get_by_slug('default'))
print "Id: {}".format(data_source.id)
开发者ID:babybits，项目名称:redash，代码行数:58，代码来源:data_sources.py
示例5: test_doesnt_leave_leftovers
def test_doesnt_leave_leftovers(self):
container = ConfigurationContainer({'a': 1, 'b': 'test', 'e': 3}, configuration_schema)
new_config = container.to_dict(mask_secrets=True)
new_config.pop('e')
container.update(new_config)
self.assertEqual(container['a'], 1)
self.assertEqual('test', container['b'])
self.assertNotIn('e', container)
开发者ID:babybits，项目名称:redash，代码行数:9，代码来源:test_configuration.py
示例6: TestConfigurationUpdate
class TestConfigurationUpdate(TestCase):
def setUp(self):
self.config = {'a': 1, 'b': 'test'}
self.container = ConfigurationContainer(self.config, configuration_schema)
def test_rejects_invalid_new_config(self):
self.assertRaises(ValidationError, lambda: self.container.update({'c': 3}))
def test_fails_if_no_schema_set(self):
self.container.set_schema(None)
self.assertRaises(RuntimeError, lambda: self.container.update({'c': 3}))
def test_ignores_secret_placehodler(self):
self.container.update(self.container.to_dict(mask_secrets=True))
self.assertEqual(self.container['b'], self.config['b'])
def test_updates_secret(self):
new_config = {'a': 2, 'b': 'new'}
self.container.update(new_config)
self.assertDictEqual(self.container._config, new_config)
def test_doesnt_leave_leftovers(self):
container = ConfigurationContainer({'a': 1, 'b': 'test', 'e': 3}, configuration_schema)
new_config = container.to_dict(mask_secrets=True)
new_config.pop('e')
container.update(new_config)
self.assertEqual(container['a'], 1)
self.assertEqual('test', container['b'])
self.assertNotIn('e', container)
开发者ID:babybits，项目名称:redash，代码行数:30，代码来源:test_configuration.py
示例7: TestConfigurationToJson
class TestConfigurationToJson(TestCase):
def setUp(self):
self.config = {'a': 1, 'b': 'test'}
self.container = ConfigurationContainer(self.config, configuration_schema)
def test_returns_plain_dict(self):
self.assertDictEqual(self.config, self.container.to_dict())
def test_raises_exception_when_no_schema_set(self):
self.container.set_schema(None)
self.assertRaises(RuntimeError, lambda: self.container.to_dict(mask_secrets=True))
def test_returns_dict_with_masked_secrets(self):
d = self.container.to_dict(mask_secrets=True)
self.assertEqual(d['a'], self.config['a'])
self.assertNotEqual(d['b'], self.config['b'])
self.assertEqual(self.config['b'], self.container['b'])
开发者ID:babybits，项目名称:redash，代码行数:19，代码来源:test_configuration.py
示例8: post
def post(self):
req = request.get_json(True)
required_fields = ("options", "name", "type")
for f in required_fields:
if f not in req:
abort(400)
schema = get_configuration_schema_for_query_runner_type(req["type"])
if schema is None:
abort(400)
config = ConfigurationContainer(req["options"], schema)
if not config.is_valid():
abort(400)
datasource = models.DataSource.create_with_group(
org=self.current_org, name=req["name"], type=req["type"], options=config
)
self.record_event({"action": "create", "object_id": datasource.id, "object_type": "datasource"})
return datasource.to_dict(all=True)
开发者ID:hudl，项目名称:redash，代码行数:21，代码来源:data_sources.py
示例9: post
def post(self):
req = request.get_json(True)
required_fields = ('options', 'name', 'type')
for f in required_fields:
if f not in req:
abort(400)
schema = get_configuration_schema_for_destination_type(req['type'])
if schema is None:
abort(400)
config = ConfigurationContainer(req['options'], schema)
if not config.is_valid():
abort(400)
destination = models.NotificationDestination(org=self.current_org,
name=req['name'],
type=req['type'],
options=config,
user=self.current_user)
destination.save()
return destination.to_dict(all=True)
开发者ID:hudl，项目名称:redash，代码行数:23，代码来源:destinations.py
示例10: test_adds_data_source_to_default_group
def test_adds_data_source_to_default_group(self):
data_source = DataSource.create_with_group(org=self.factory.org, name='test', options=ConfigurationContainer.from_json('{"dbname": "test"}'), type='pg')
self.assertIn(self.factory.org.default_group.id, data_source.groups)
开发者ID:5t111111，项目名称:redash，代码行数:3，代码来源:test_data_sources.py
示例11: setUp
def setUp(self):
self.config = {'a': 1, 'b': 'test'}
self.container = ConfigurationContainer(self.config, configuration_schema)
开发者ID:babybits，项目名称:redash，代码行数:3，代码来源:test_configuration.py
示例12: ModelFactory
user_factory = ModelFactory(redash.models.User,
name='John Doe', email=Sequence('test{}@example.com'),
groups=[2],
org=1)
org_factory = ModelFactory(redash.models.Organization,
name=Sequence("Org {}"),
slug=Sequence("org{}.example.com"),
settings={})
data_source_factory = ModelFactory(redash.models.DataSource,
name=Sequence('Test {}'),
type='pg',
options=ConfigurationContainer.from_json('{"dbname": "test"}'),
org=1)
dashboard_factory = ModelFactory(redash.models.Dashboard,
name='test', user=user_factory.create, layout='[]', org=1)
query_factory = ModelFactory(redash.models.Query,
name='New Query',
description='',
query='SELECT 1',
user=user_factory.create,
is_archived=False,
schedule=None,
data_source=data_source_factory.create,
org=1)
开发者ID:babybits，项目名称:redash，代码行数:28，代码来源:factories.py
示例13: process_result_value
def process_result_value(self, value, dialect):
return ConfigurationContainer.from_json(super(EncryptedConfiguration, self).process_result_value(value, dialect))
开发者ID:ariarijp，项目名称:redash，代码行数:2，代码来源:types.py
示例14: ModelFactory
user_factory = ModelFactory(redash.models.User,
name='John Doe', email=Sequence('test{}@example.com'),
groups=[2],
org=1)
org_factory = ModelFactory(redash.models.Organization,
name=Sequence("Org {}"),
slug=Sequence("org{}.example.com"),
settings={})
data_source_factory = ModelFactory(redash.models.DataSource,
name=Sequence('Test {}'),
type='pg',
# If we don't use lambda here it will reuse the same options between tests:
options=lambda: ConfigurationContainer.from_json('{"dbname": "test"}'),
org=1)
dashboard_factory = ModelFactory(redash.models.Dashboard,
name='test', user=user_factory.create, layout='[]', org=1)
api_key_factory = ModelFactory(redash.models.ApiKey,
object=dashboard_factory.create)
query_factory = ModelFactory(redash.models.Query,
name='New Query',
description='',
query='SELECT 1',
user=user_factory.create,
is_archived=False,
schedule=None,
开发者ID:5t111111，项目名称:redash，代码行数:30，代码来源:factories.py
示例15: python_value
def python_value(self, value):
return ConfigurationContainer.from_json(value)
开发者ID:Drunkar，项目名称:redash，代码行数:2，代码来源:models.py
示例16: test_works_for_schema_without_secret
def test_works_for_schema_without_secret(self):
secretless = configuration_schema.copy()
secretless.pop('secret')
container = ConfigurationContainer({'a': 1, 'b': 'test', 'e': 3}, secretless)
container.update({'a': 2})
self.assertEqual(container['a'], 2)
开发者ID:13768324554，项目名称:redash，代码行数:6，代码来源:test_configuration.py
注：本文中的redash.utils.configuration.ConfigurationContainer类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论