Background task blocks the request when I trigger the endpoint from fastapi api docs

[Tianduo16]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

async def transfer_apify_results_in_batch(metadata: ApifyWebhookPayload) -> int:
    # get crawler run job overview
    log = structlog.get_logger(job_id=metadata.eventData.actorRunId)
    apify_token = env.str("APIFY_TOKEN")
    client = httpx.AsyncClient(timeout=60, http2=True)
    response = await client.get(
        f"https://api.apify.com/v2/actor-runs/{metadata.eventData.actorRunId}/dataset",
        headers={
            "Authorization": f"Bearer {apify_token}",
        },
    )
    response.raise_for_status()
    dataset_overview = response.json()
    total: int | None = dataset_overview.get("data", {}).get("cleanItemCount")
    log.info(f"total {total} dataset in job")
    if not total:
        raise Exception("No data found in the dataset")
    # transfer the dataset in batch
    batch_size = min(
        1000, env.int("APIFY_CRAWL_RESULT_BATCH_SIZE", 500)
    )  # https://docs.apify.com/api/v2#/reference/datasets/dataset-collection/get-list-of-datasets max value of limit is 1000  # noqa
    offset = metadata.offset
    knowledge_source_sub_path = (
        f"knowledge-source-{metadata.knowledgeSourceId}" if metadata.knowledgeSourceId else "unknown"
    )

    bucket = GCSClient.get_storage_client().bucket(env.str("GCS_BUCKET_NAME"))
    while offset < total:
        log.info(f"transferring dataset from {offset} to {offset + batch_size}")
        response = await client.get(
            f"https://api.apify.com/v2/actor-runs/{metadata.eventData.actorRunId}/dataset/items?format=json&limit={batch_size}&offset={offset}",
            headers={
                "Authorization": f"Bearer {apify_token}",
            },
        )
        response.raise_for_status()
        datasets = response.content
        # Save the dataset to GCS bucket specified by GCS_BUCKET_NAME and GCS_PREFIX as a file

        blob = bucket.blob(
            f"{GCS_PREFIX}/{metadata.eventData.actorRunId}-{offset}-{batch_size}.json"
        )
        # Once blob.metadata is set, it can't be changed directly.
        object_metadata = {
            "tenant_id": metadata.tenantId,
            "actor_run_id": metadata.eventData.actorRunId,
        }
        # Add site_ids to metadata if it exists
        if metadata.siteIds:
            object_metadata["site_ids"] = json.dumps([str(u) for u in metadata.siteIds])
        blob.metadata = object_metadata
        blob.upload_from_string(
            data=datasets,
            content_type="application/json",
        )
        offset += batch_size
    await client.aclose()
    return total


@app.post(
    "/crawl/apify-webhook/run-succeeded",
    summary="Webhook handler for Apify actor run succeeded event",
    tags=["crawl"],
)
async def apify_webhook_run_succeeded(payload: ApifyWebhookPayload, background_tasks: BackgroundTasks) -> str:
    logger.info("apify_webhook_run_succeeded", payload=payload)
    background_tasks.add_task(transfer_apify_results_in_batch, payload)
    return "OK"

Description

I have this endpoint as a webhook for Apify crawler to callback when a crawl run job is finished.
I found a weird thing about this:

If I trigger this endpoint from browser, which is the fastapi api docs, the request will wait for the background task to finish and return the status code.

If I trigger the endpoint using a curl, it can return immediately and run the background task in the background.

Operating System

Linux, macOS

Operating System Details

I've observed this in our server which is running on cloud run on linux, I can reproduce this very easily in my macos

FastAPI Version

0.106.0

Pydantic Version

2.7.2

Python Version

Python 3.11.9

Additional Context

No response

[mattmess1221]

Your GCS bucket upload is a blocking call, which is causing the server to freeze while it uploads. You should wrap the upload function with asyncio.to_thread() (3.9+) or asyncio.get_current_loop().run_in_executor() (3.8).

import asyncio

# 3.9+
await asyncio.to_thread(
    blob.upload_from_string,
    data=datasets,
    content_type="application/json",
)

# 3.8
from functools import partial

loop = asyncio.get_running_loop()
await loop.run_in_executor(None, partial(
    blob.upload_from_string,
    data=datasets,
    content_type="application/json",
))

See the google-cloud-storage samples.
https://github.com/googleapis/python-storage/blob/main/samples/snippets/storage_async_upload.py

[Tianduo16]

Hey thanks for taking a look at this issue.

The server is not frozen, it's just the request is blocked by backgroundtask, it's different thing.

I found this discussion Kludex/starlette#919 might be related. I tried commented out the middlewares and is works normally. But I did not find a solution yet.

[mattmess1221]

Are you sure the server isn't frozen? Do 2 curl requests and time each one. If the second request takes longer, then the server was blocked.

[Tianduo16]

I got your point running the gcs upload in sync way instead of the async way. Thanks for providing the example! I will try that.

But this is not my problem so far, the request is blocked by the background task not the upload. I don't need a second request to be blocked.

[Tianduo16]

I updated the version to 0.111.0 and this is fixed