Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def unset(updates):
track_cli('config-unset')
local_config = ConfigParser()
local_config_file = get_local_config_file()
local_config.read(local_config_file, encoding='utf-8')
try:
for update in updates:
if '.' in update:
sec, opt = update.split('.')
else:
sec = 'core' # default section
opt = update
if not local_config.has_section(sec):
local_config.add_section(sec)
local_config.remove_option(sec.strip(), opt.strip())
'dependencies."; }} &&'
'bentoml serve_gunicorn {bundle_path} -p {port} -w {workers} '
'--timeout {timeout}'.format(
env_name=env_name,
env_file=os.path.join(bundle_path, 'environment.yml'),
bundle_path=bundle_path,
port=port,
workers=workers,
timeout=timeout,
pip_req=pip_req,
),
shell=True,
)
return
track_cli('serve_gunicorn')
from bentoml.server.gunicorn_server import GunicornBentoServer
gunicorn_app = GunicornBentoServer(bundle_path, port, workers, timeout)
gunicorn_app.run()
'eval "$(conda shell.bash hook)" && '
'conda activate {env_name} && '
'{{ [ -f {pip_req} ] && pip install -r {pip_req} || echo "no pip '
'dependencies."; }} &&'
'bentoml serve {bundle_path} --port {port}'.format(
env_name=env_name,
env_file=os.path.join(bundle_path, 'environment.yml'),
bundle_path=bundle_path,
port=port,
pip_req=pip_req,
),
shell=True,
)
return
track_cli('serve')
bento_service = load(bundle_path)
server = BentoAPIServer(bento_service, port=port)
server.start()
def describe(name, output, namespace):
track_cli('deploy-describe')
yatai_service = get_yatai_service()
result = describe_deployment(namespace, name, yatai_service)
if result.status.status_code != status_pb2.Status.OK:
_echo(
'Failed to describe deployment {name}. {error_code}:'
'{error_message}'.format(
name=name,
error_code=status_pb2.Status.Code.Name(result.status.status_code),
error_message=result.status.error_message,
),
CLI_COLOR_ERROR,
)
else:
get_result = get_deployment(namespace, name)
if get_result.status.status_code != status_pb2.Status.OK:
def view_effective():
track_cli('config-view-effective')
bentoml_config().write(sys.stdout)
return
def list_deployments_cli(output, limit, filters, labels, namespace, all_namespaces):
track_cli('deploy-list')
yatai_service = get_yatai_service()
result = list_deployments(
limit=limit,
filters=filters,
labels=parse_key_value_pairs(labels),
namespace=namespace,
is_all_namespaces=all_namespaces,
yatai_service=yatai_service,
)
if result.status.status_code != status_pb2.Status.OK:
_echo(
'Failed to list deployments. {error_code}:{error_message}'.format(
error_code=status_pb2.Status.Code.Name(result.status.status_code),
error_message=result.status.error_message,
),
'conda activate {env_name} && '
'{{ [ -f {pip_req} ] && pip install -r {pip_req} || echo "no pip '
'dependencies."; }} &&'
'bentoml {api_name} {bundle_path} {args}'.format(
env_name=env_name,
env_file=env_path,
bundle_path=bundle_path,
api_name=api_name,
args=' '.join(map(escape_shell_params, ctx.args)),
pip_req=pip_req,
),
shell=True,
)
return
track_cli('run')
api = load_bento_service_api(bundle_path, api_name)
api.handle_cli(ctx.args)
def get(name, output, namespace):
track_cli('deploy-get')
yatai_service = get_yatai_service()
result = get_deployment(namespace, name, yatai_service)
if result.status.status_code != status_pb2.Status.OK:
_echo(
'Failed to get deployment {name}. code: {error_code}, message: '
'{error_message}'.format(
name=name,
error_code=status_pb2.Status.Code.Name(result.status.status_code),
error_message=result.status.error_message,
),
CLI_COLOR_ERROR,
)
else:
_print_deployment_info(result.deployment, output)