Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if not arguments.command:
# default to serve command
arguments.command = "serve"
if arguments.command not in _commands:
return print(
"""Invalid command "{}".
Available commands:
{}\n\n""".format(
arguments.command, "\n - ".join(c for c in _commands.keys())
)
)
app_settings["_command"] = arguments.command
Command = resolve_dotted_name(_commands[arguments.command])
if Command is None:
return print(
"Could not resolve command {}:{}".format(arguments.command, _commands[arguments.command])
)
app_settings["__run_command__"] = arguments.command
# finally, run it...
command = Command()
command.run_command()
partition_object = resolve_dotted_name(dbconfig["partition"])
dbconfig.update(
{"dsn": dsn, "name": key, "partition": partition_object, "pool_size": dbconfig.get("pool_size", 13)}
)
connection_options = _get_connection_options(dbconfig)
aps = storage_factory(**dbconfig)
if loop is not None:
await aps.initialize(loop=loop, **connection_options)
else:
await aps.initialize(**connection_options)
if "transaction_manager" in dbconfig:
transaction_manager = resolve_dotted_name(dbconfig["transaction_manager"])
else:
transaction_manager = TransactionManager
db = Database(key, aps, transaction_manager)
await db.initialize()
return db
def load_behavior(_context, behavior):
conf = behavior['config']
klass = resolve_dotted_name(behavior['klass'])
factory = conf.get('factory') or klass
real_factory = resolve_dotted_name(factory)
if IInterface.providedBy(real_factory):
# create concret class to register for behavior
schema = real_factory
from guillotina.behaviors.instance import AnnotationBehavior
class real_factory(AnnotationBehavior):
__annotations_data_key__ = conf.get('data_key', 'default')
auto_serialize = conf.get('auto_serialize', True)
else:
schema = resolve_dotted_name(conf['provides'])
classImplements(real_factory, schema)
name = conf.get('name')
name_only = conf.get('name_only', False)
title = conf.get('title', '')
for_ = resolve_dotted_name(conf.get('for_'))
marker = resolve_dotted_name(conf.get('marker'))
if marker is None and real_factory is None:
marker = schema
if marker is not None and real_factory is None and marker is not schema:
raise ConfigurationError(
u"You cannot specify a different 'marker' and 'provides' if "
u"there is no adapter factory for the provided interface."
)
def load_service(_context, service):
# prevent circular import
from guillotina.security.utils import protect_view
service_conf = service['config']
factory = resolve_dotted_name(service['klass'])
permission = service_conf.get(
'permission', app_settings.get('default_permission', None))
protect_view(factory, permission)
method = service_conf.get('method', 'GET')
default_layer = resolve_dotted_name(
app_settings.get('default_layer', IDefaultLayer))
layer = service_conf.get('layer', default_layer)
name = service_conf.get('name', '')
content = service_conf.get('context', Interface)
logger.debug('Defining adapter for ' # noqa
'{0:s} {1:s} {2:s} to {3:s} name {4:s}'.format(
content.__identifier__,
app_settings['http_methods'][method].__identifier__,
def get_configurations(module_name, type_=None, excluded=None):
results = []
for reg_type, registration in _registered_configurations:
if type_ is not None and reg_type != type_:
continue
config = registration['config']
module = config.get('module', registration.get('klass'))
normalized_name = get_module_dotted_name(resolve_dotted_name(module))
if (normalized_name + '.').startswith(module_name + '.'):
valid = True
for excluded_module in excluded or []:
if (normalized_name + '.').startswith(excluded_module + '.'):
valid = False
break
if valid:
results.append((reg_type, registration))
return results
def optimize_settings(settings):
"""
pre-render settings that come in as strings but are used by the app
"""
for name in _dotted_name_settings:
if name not in settings:
continue
val = settings[name]
if isinstance(val, str):
settings[name] = resolve_dotted_name(val)
elif isinstance(val, list):
new_val = []
for v in val:
if isinstance(v, str):
v = resolve_dotted_name(v)
new_val.append(v)
settings[name] = resolve_dotted_name(new_val)
def load_commands(module_name, commands):
module = resolve_dotted_name(module_name)
if hasattr(module, "app_settings") and app_settings != module.app_settings:
commands.update(module.app_settings.get("commands", {}))
for dependency in module.app_settings.get("applications") or []:
load_commands(dependency, commands)
def load_utility(_context, _utility):
conf = _utility["config"]
if "factory" in conf:
conf["factory"] = resolve_dotted_name(conf["factory"])
elif "component" in conf:
conf["component"] = resolve_dotted_name(conf["component"])
else:
# use provided klass
klass = _utility["klass"]
if isinstance(klass, type):
# is a class type, use factory setting
conf["factory"] = klass
else:
# not a factory
conf["component"] = klass
component.utility(_context, **conf)