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_file(config):
ns = 'banana'
config_path = create_yaml_config(namespace=ns)
with patch_env({mlconf.env_file_key: config_path}):
mlconf.config.reload()
assert config.namespace == ns, 'not populated from file'
def config():
old = mlconf.config
mlconf.config = mlconf.Config(mlconf.default_config)
mlconf._loaded = False
yield mlconf.config
mlconf.config = old
mlconf._loaded = False
if kfp:
print('Runtime:')
pprint(runtime)
func = new_function(runtime=runtime)
elif func_url.startswith('db://'):
func_url = func_url[5:]
func = import_function(func_url)
elif func_url:
func_url = 'function.yaml' if func_url == '.' else func_url
func = import_function(func_url)
else:
print('please specify the function path or url')
exit(1)
meta = func.metadata
meta.project = project or meta.project or mlconf.default_project
meta.name = name or meta.name
meta.tag = tag or meta.tag
b = func.spec.build
if func.kind not in ['', 'local']:
b.base_image = base_image or b.base_image
b.commands = list(command) or b.commands
b.image = image or b.image
b.secret = secret_name or b.secret
if source.endswith('.py'):
if not path.isfile(source):
print('source file doesnt exist ({})'.format(source))
exit(1)
with open(source) as fp:
body = fp.read()
def get_or_set_dburl(default=''):
if not config.dbpath and default:
config.dbpath = default
environ['MLRUN_DBPATH'] = default
return config.dbpath
def init_db(db_session: Session) -> None:
if config.httpdb.db_type != "filedb":
Base.metadata.create_all(bind=get_engine())
def main():
app.run(
host='0.0.0.0',
port=config.httpdb.port,
debug=config.httpdb.debug,
)
raise ValueError(
'local docker registry is not defined, set '
'DEFAULT_DOCKER_REGISTRY/SECRET env vars'
)
dest = '{}/{}'.format(registry, dest)
if isinstance(requirements, list):
requirements_list = requirements
requirements_path = 'requirements.txt'
if source:
raise ValueError('requirements list only works with inline code')
else:
requirements_list = None
requirements_path = requirements
base_image = base_image or config.default_image
if with_mlrun:
commands = commands or []
commands.append('pip install {}'.format(config.package_path))
if not inline_code and not source and not commands:
logger.info('skipping build, nothing to add')
return 'skipped'
context = '/context'
to_mount = False
src_dir = '.'
v3io = (
source.startswith('v3io://') or source.startswith('v3ios://')
if source
else None
)
def get_obj_path(schema, path):
if schema:
return schema + '://' + path
elif path.startswith('/User/'):
user = environ.get('V3IO_USERNAME', 'admin')
return 'v3io:///users/' + user + path[5:]
elif config.httpdb.data_volume and \
path.startswith(config.httpdb.data_volume):
if config.httpdb.real_path:
path = config.httpdb.real_path + \
path[len(config.httpdb.data_volume)-1:]
return path
return None
def get_function(self, session, name, project="", tag="", hash_key=""):
project = project or config.default_project
query = self._query(session, Function, name=name, project=project)
computed_tag = tag or "latest"
tag_function_uid = None
if not tag and hash_key:
uid = hash_key
else:
tag_function_uid = self._resolve_tag_function_uid(session, Function, project, name, computed_tag)
uid = tag_function_uid
if uid:
query = query.filter(Function.uid == uid)
obj = query.one_or_none()
if obj:
function = obj.struct
# If queried by hash key remove status
if hash_key: