Create a web wrapper for job queue, submission, polling, & results

This simple tutorial shows you how to create an API endpoint that you can use to poll the status of your request.

Let’s first import modal and define an App.

import time

import modal

app = modal.App("example-web-job-queue-wrapper")

Next, we’ll create a dummy backend service, in reality you may plug an a LLM or Diffusion model here. We’ll add artificial delays to simulate a cold boot and a long-running tasks.

@app.cls()
class BackendService:
    @modal.enter()
    def enter(self):
        print("begin cold booting")
        time.sleep(10)
        print("end cold booting")

    @modal.method()
    def run(self, input_val: str):
        print(f"begin run with {input_val}")
        time.sleep(5)
        print(f"end run with {input_val}")
        return input_val[::-1]  # reverse the string

Then, we can define a Web Function that will submit a request to the backend service as well as other API routes for polling or retrieving results.

To submit jobs asynchronously, we can use ‘spawn’, which return a FunctionCall object that represents the submitted job.

Then we can poll for the job’s status by calling .get(timeout=0) on the FunctionCall, which returns immediately with the result if it’s ready and otherwise raises a TimeoutError while the job is still running.

@app.function(
    image=modal.Image.debian_slim().uv_pip_install("fastapi[standard]==0.116.0")
)
@modal.asgi_app()
@modal.concurrent(max_inputs=100)
def gateway():
    from fastapi import FastAPI, Request
    from modal.exception import OutputExpiredError

    web_app = FastAPI()

    service = BackendService()

    @web_app.post("/run")
    async def submit(request: Request):
        """Asynchronously submit a request to the backend service."""
        input_val = (await request.json())["input_val"]
        fc = service.run.spawn(input_val)
        return {"request_id": fc.object_id}

    @web_app.get("/requests/{request_id}/status")
    async def status(request_id: str):
        """Get the status of the request by polling for a result without blocking."""
        fc = modal.FunctionCall.from_id(request_id)
        try:
            await fc.get.aio(timeout=0)
            return {"status": "SUCCESS"}
        except OutputExpiredError:
            return {"status": "EXPIRED"}
        except TimeoutError:
            return {"status": "PENDING"}
        except Exception:
            return {"status": "FAILURE"}

    @web_app.get("/requests/{request_id}")
    async def result(request_id: str):
        fc = modal.FunctionCall.from_id(request_id)
        return {"response": await fc.get.aio()}

    return web_app

To test this you can do:

modal serve web_job_queue_wrapper.py

Or run the test locally:

modal run web_job_queue_wrapper.py::test_polling

@app.local_entrypoint()
def test_polling():
    """Test the polling job queue by submitting a request and polling for results."""
    import json
    import urllib.parse
    import urllib.request

    # Get the deployed URL
    url = gateway.get_web_url()
    print(f"URL: {url}")

    # Submit request
    print("submitting request")
    data = json.dumps({"input_val": "Hello, world!"}).encode("utf-8")
    headers = {"Content-Type": "application/json"}
    req = urllib.request.Request(
        f"{url}/run", data=data, headers=headers, method="POST"
    )

    try:
        with urllib.request.urlopen(req) as response:
            result = json.loads(response.read().decode("utf-8"))
            request_id = result["request_id"]
            print(f"got request id: {request_id}, polling status")
    except Exception as e:
        print(f"Failed to submit request: {e}")
        return

    # Poll for status
    while True:
        try:
            with urllib.request.urlopen(
                f"{url}/requests/{request_id}/status"
            ) as response:
                data = json.loads(response.read().decode("utf-8"))
                if data["status"] == "SUCCESS":
                    print("request completed successfully")
                    break
                else:
                    print(f"request result is {data['status']}")
        except Exception as e:
            print(f"poll failed: {e}")
        time.sleep(1)

    # Retrieve result
    print("retrieving result")
    try:
        with urllib.request.urlopen(f"{url}/requests/{request_id}") as response:
            result = json.loads(response.read().decode("utf-8"))
            print(f"result is {result}")
            print("done")
    except Exception as e:
        print(f"Failed to retrieve result: {e}")

Create a web wrapper for job queue, submission, polling, & results

Try this on Modal!