#How to upload a file using a upload api in gpu serverless?
This is my current code, there is a separate fastapi running.
#⚡｜serverless
automatic_session = requests.Session()
retries = Retry(total=10, backoff_factor=0.1, status_forcelist=[502, 503, 504])
automatic_session.mount('http://', HTTPAdapter(max_retries=retries))
def run_inference(params):
config = {
"baseurl": "http://127.0.0.1:8080",
"api": {
"health_check": ("GET", "/health_check"),
"upload": ("POST", "/upload"),
},
"timeout": 600
}
api_name = params["api_name"]
if api_name in config["api"]:
api_config = config["api"][api_name]
else:
raise Exception("Method '%s' not yet implemented")
api_verb = api_config[0]
api_path = api_config[1]
response = {}
if api_verb == "GET":
response = automatic_session.get(
url='%s%s' % (config["baseurl"], api_path),
timeout=config["timeout"])
elif api_verb == "POST":
if "upload" in api_path:
print("Inside upload rp")
# For "upload" endpoints, use form data
files = {'file': ('filename', params['file'].read())} # Assuming the file is passed in params
response = automatic_session.post(
url='%s%s' % (config["baseurl"], api_path),
files=files,
timeout=config["timeout"])
else:
# For other POST requests, use JSON
response = automatic_session.post(
url='%s%s' % (config["baseurl"], api_path),
json=params,
timeout=config["timeout"])
return response.json()
def handler(event):
json = run_inference(event["input"])
return json
if __name__ == "__main__":
runpod.serverless.start({"handler": handler})
automatic_session = requests.Session()
retries = Retry(total=10, backoff_factor=0.1, status_forcelist=[502, 503, 504])
automatic_session.mount('http://', HTTPAdapter(max_retries=retries))
def run_inference(params):
config = {
"baseurl": "http://127.0.0.1:8080",
"api": {
"health_check": ("GET", "/health_check"),
"upload": ("POST", "/upload"),
},
"timeout": 600
}
api_name = params["api_name"]
if api_name in config["api"]:
api_config = config["api"][api_name]
else:
raise Exception("Method '%s' not yet implemented")
api_verb = api_config[0]
api_path = api_config[1]
response = {}
if api_verb == "GET":
response = automatic_session.get(
url='%s%s' % (config["baseurl"], api_path),
timeout=config["timeout"])
elif api_verb == "POST":
if "upload" in api_path:
print("Inside upload rp")
# For "upload" endpoints, use form data
files = {'file': ('filename', params['file'].read())} # Assuming the file is passed in params
response = automatic_session.post(
url='%s%s' % (config["baseurl"], api_path),
files=files,
timeout=config["timeout"])
else:
# For other POST requests, use JSON
response = automatic_session.post(
url='%s%s' % (config["baseurl"], api_path),
json=params,
timeout=config["timeout"])
return response.json()
def handler(event):
json = run_inference(event["input"])
return json
if __name__ == "__main__":
runpod.serverless.start({"handler": handler})
2 Replies