fastapi_dummy / main.py
youngtsai's picture
with NamedTemporaryFile(delete=False) as tmp_file:
94ee50b
from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.responses import JSONResponse
import requests
import os
from tempfile import NamedTemporaryFile
app = FastAPI()
PASSWORD = os.getenv("PASSWORD") # Read password from environment variable
ELSA_API_TOKEN = os.getenv("ELSA_API_TOKEN") # Make sure to set this environment variable
def verify_password(input_password: str):
return input_password == PASSWORD
@app.get("/")
def read_root():
return {"Hello": "World!"}
@app.post("/score_audio_plus/")
async def score_audio_plus(
password: str = Form(...),
audio: UploadFile = File(...),
api_plan: str = Form(...),
return_json: bool = Form(...)
):
"""
Call the ELSA 'score_audio_plus' endpoint with the provided audio file.
"""
if not audio:
return JSONResponse(content={"message": "No audio file provided. Please upload an audio file."}, status_code=400)
if not verify_password(password):
raise HTTPException(status_code=401, detail="Unauthorized")
url = "https://api.elsanow.io/api/v1/score_audio_plus"
headers = {"Authorization": f"ELSA {ELSA_API_TOKEN}"}
# Write file content to a temporary file
with NamedTemporaryFile(delete=False) as tmp_file:
tmp_file.write(await audio.read())
tmp_file_name = tmp_file.name
files = {
'audio_file': (os.path.basename(audio.filename), open(tmp_file_name, "rb"), 'audio/wav')
}
data = {
'api_plan': api_plan,
'return_json': "true" if return_json else "false"
}
# Send POST request
response = requests.post(url, files=files, data=data, headers=headers)
os.unlink(tmp_file_name) # Remove the temporary file
if response.status_code == 200:
# Successfully made the request
return JSONResponse(content=response.json(), status_code=200)
else:
# Error handling
return JSONResponse(content={"error": response.text}, status_code=response.status_code)