Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def generate_sagemaker_deployment_pb():
test_deployment_pb = Deployment(
name=TEST_DEPLOYMENT_NAME, namespace=TEST_DEPLOYMENT_NAMESPACE
)
test_deployment_pb.spec.bento_name = TEST_DEPLOYMENT_BENTO_NAME
test_deployment_pb.spec.bento_version = TEST_DEPLOYMENT_BENTO_VERSION
test_deployment_pb.spec.sagemaker_operator_config.api_name = TEST_BENTO_API_NAME
test_deployment_pb.spec.sagemaker_operator_config.region = TEST_AWS_REGION
test_deployment_pb.spec.sagemaker_operator_config.instance_count = (
TEST_DEPLOYMENT_INSTANCE_COUNT
)
test_deployment_pb.spec.sagemaker_operator_config.instance_type = (
TEST_DEPLOYMENT_INSTANCE_TYPE
)
test_deployment_pb.spec.operator = DeploymentSpec.AWS_SAGEMAKER
return test_deployment_pb
def generate_lambda_deployment_pb():
test_deployment_pb = Deployment(name='test_aws_lambda', namespace='test-namespace')
test_deployment_pb.spec.bento_name = 'bento_name'
test_deployment_pb.spec.bento_version = 'v1.0.0'
# DeploymentSpec.DeploymentOperator.AWS_LAMBDA
test_deployment_pb.spec.operator = 3
test_deployment_pb.spec.aws_lambda_operator_config.region = 'us-west-2'
test_deployment_pb.spec.aws_lambda_operator_config.api_name = 'predict'
return test_deployment_pb
def generate_lambda_deployment_pb():
test_deployment_pb = Deployment(name='test_aws_lambda', namespace='test-namespace')
test_deployment_pb.spec.bento_name = 'bento_name'
test_deployment_pb.spec.bento_version = 'v1.0.0'
# DeploymentSpec.DeploymentOperator.AWS_LAMBDA
test_deployment_pb.spec.operator = 3
test_deployment_pb.spec.aws_lambda_operator_config.region = 'us-west-2'
test_deployment_pb.spec.aws_lambda_operator_config.api_name = 'predict'
test_deployment_pb.spec.aws_lambda_operator_config.memory_size = 3008
test_deployment_pb.spec.aws_lambda_operator_config.timeout = 6
return test_deployment_pb
def test_validate_aws_lambda_schema():
test_pb = Deployment(name='test_deployment_name', namespace='namespace')
test_pb.spec.bento_name = 'bento_name'
test_pb.spec.bento_version = 'bento_version'
test_pb.spec.operator = DeploymentSpec.DeploymentOperator.Value('AWS_LAMBDA')
test_pb.spec.aws_lambda_operator_config.api_name = 'api_name'
test_pb.spec.aws_lambda_operator_config.region = 'us-west-2'
test_pb.spec.aws_lambda_operator_config.timeout = 100
test_pb.spec.aws_lambda_operator_config.memory_size = 128
result = validate_deployment_pb_schema(test_pb)
assert result is None
test_pb.spec.aws_lambda_operator_config.timeout = 1000
test_pb.spec.aws_lambda_operator_config.memory_size = 129
failed_memory_test = validate_deployment_pb_schema(test_pb)
print(failed_memory_test)
aws_spec_fail_msg = failed_memory_test['spec'][0]['aws_lambda_operator_config'][0]
serverless_project_dir,
aws_config.region,
# BentoML deployment namespace is mapping to serverless `stage`
# concept
stage=deployment_pb.namespace,
)
logger.info(
'Installing additional packages: serverless-python-requirements'
)
install_serverless_plugin(
"serverless-python-requirements", serverless_project_dir
)
logger.info('Deploying to AWS Lambda')
call_serverless_command(["deploy"], serverless_project_dir)
res_deployment_pb = Deployment(state=DeploymentState())
res_deployment_pb.CopyFrom(deployment_pb)
state = self.describe(res_deployment_pb, yatai_service).state
res_deployment_pb.state.CopyFrom(state)
return ApplyDeploymentResponse(status=Status.OK(), deployment=res_deployment_pb)