Skip to main content

​
Handling cancellations

A cancellation is received in the same endpoint that is processing the request with a /cancel suffix. So for the endpoint /predict you will receive a request to /predict/cancel. Request id are passed in the x-fal-request-id header for both the request and the cancellation.

​
Example

import asyncio

import fal
from fal.exceptions import RequestCancelledException
import fastapi
from pydantic import BaseModel


class HelloInput(BaseModel):
    name: str


class MyApp(fal.App):
    _tasks: dict[str, asyncio.Task] = {}

    async def _predict(self, input: HelloInput):
        await asyncio.sleep(10)
        return "Hello, " + input.name

    @fal.endpoint("/predict")
    async def predict(
        self,
        input: HelloInput,
        x_fal_request_id: str | None = fastapi.Header(None),
    ):
        if not x_fal_request_id:
            raise ValueError("x-fal-request-id is required")

        task = asyncio.create_task(self._predict(input))
        self._tasks[x_fal_request_id] = task
        try:
            return await task
        except asyncio.CancelledError:
            # Run any cancellation logic here to leave the app in good state
            print("Request cancelled")
            raise RequestCancelledException("Request cancelled")
        finally:
            self._tasks.pop(x_fal_request_id)

    @fal.endpoint("/predict/cancel")
    async def predict_cancel(
        self,
        x_fal_request_id: str | None = fastapi.Header(None),
    ):
        if not x_fal_request_id:
            raise ValueError("x-fal-request-id is required")

        if x_fal_request_id not in self._tasks:
            raise ValueError("No task to cancel")

        if self._tasks[x_fal_request_id].done():
            raise ValueError("Task is already done")

        print(f"Cancelling request {x_fal_request_id}")
        self._tasks[x_fal_request_id].cancel()