Call the SageMaker endpoint from local

local_call_endpoint.py

import pandas as pd
import json
import boto3

# app name and region are global variables
global app_name
global region
app_name = 'NaiveBayesTest'
region = "eu-west-1"

def check_status(app_name):
    r""" Call the sagemaker endpoint and return the 
    request status
    Parameter
    ---------
    app_name: str, name of the endpoint
    
    Return 
    ------
    endpoint_status: str, status of the endpoint call
    """
    sage_client = boto3.client('sagemaker', region_name=region)
    endpoint_description = sage_client.describe_endpoint(EndpointName=app_name)
    endpoint_status = endpoint_description['EndpointStatus']
    return endpoint_status


def query_endpoint(app_name, input_json):
    r""" Invoke the SageMaker endpoint and send the 
    input request to be processed 

    Parameter
    ---------
    app_name: str, name of the endpoint
    input_json: str, input json tweet to be processed
    
    Return 
    ------
    preds: 1/0 prediction 
    """
    client = boto3.session.Session().client('sagemaker-runtime', region)

    response = client.invoke_endpoint(
        EndpointName = app_name,
        Body = input_json,
        ContentType = 'application/json; format=pandas-split',
        )

    preds = response['Body'].read().decode('ascii')
    preds = json.loads(preds)
    print('Received response: {}'.format(preds))
    return preds


# Check endpoint status
print('Application status is {}'.format(check_status(app_name)))

# Prepare to give for predictions
tweet = "ok so saying I'm freeeeee is a little premature, study still require...however in my best putting off mode, will start tomorrow"
# convert the input in a suitable format for SageMaker endpoint
input_df = pd.DataFrame({"text":tweet}, index=[0]).to_json(orient="split")
# return predictions
predictions = query_endpoint(app_name=app_name, input_json=input_df)