Tiktok-Talent-Info/endpoints/text.py

import asyncio
from fastapi import Form
from fastapi.responses import JSONResponse
from asyncio import to_thread
from pipeline_setup import pipe

# api
async def text_query(question: str = Form(...)):
    """
    API endpoint to process text input with the user's query.
    """
    try:
        print("starting text querying...")
        response = await to_thread(pipe, question) 
        return JSONResponse({"query": question, "response": response.text})
    except Exception as e:
        return JSONResponse({"query": question, "error": str(e)})

# gradio
# async def text_query(question: str = Form(...)):
#     """
#     API endpoint to process text input with the user's query.
#     """
#     try:
#         print("Processing in text.py...")
#         response = await to_thread(pipe, question)
#         return {"query": question, "response": response.text}
#     except Exception as e:
#         return {"query": question, "error": str(e)}

# celery
# def text_query(question: str = Form(...)):
#     """
#     API endpoint to process text input with the user's query.
#     """
#     print("Testing....")
#     try:
#         print("Processing in text.py...")
#         response = pipe(question)  # Call pipe synchronously
#         return {"query": question, "response": response.text}
#     except Exception as e:
#         return {"query": question, "error": str(e)}