Concurrently start function and return early

user_1061572 0
Viewed 604

I need to return a response from my FastAPI path operation, but before this I want to send a slow request and I don't need to wait for result of that request, just log errors if there are any. Can I do this by means of Python and FastAPI? I would not like to add Celery to the project.

Here is what I have so far, but it runs synchronously:

import asyncio
import requests


async def slow_request(data):
    url = 'https://external.service'
    response = requests.post(
        url=url,
        json=data,
        headers={'Auth-Header': settings.API_TOKEN}
    )
    if not response.status_code == 200:
        logger.error('response:', response.status_code)
        logger.error('data', data)


@router.post('/order/')
async def handle_order(order: Order):
    json_data = {
        'order': order
    }
    
    task = asyncio.create_task(
        slow_request(json_data)
    )
    await task

    return {'body': {'message': 'success'}}
fastapi python-asyncio python
2 Answers

OK, if nobody wants to post an answer here are the solutions:

Solution #1

We can just remove await task line as alex_noname suggested. It will work because create_task schedules task and we are no longer awaiting for its completion.

@router.post('/order/')
async def handle_order(order: Order):
    json_data = {
        'order': order
    }
    
    task = asyncio.create_task(
        slow_request(json_data)
    )

    return {'body': {'message': 'success'}}

Solution #2

I ended up with BackgroundTasks as HTF suggested as I'm already using FastAPI anyway, and this solution seems more neat to me.

@router.post('/order/')
async def handle_order(order: Order, background_tasks: BackgroundTasks):
    json_data = {
        'order': order
    }
    
    background_tasks.add_task(slow_request, json_data)

    return {'body': {'message': 'success'}}

This even works without async before def slow_request(data):

user_10615720

The problem is really two-part

  • the requests library is synchronous, so requests.post(...) will block the event loop until completed
  • you don't need the result of the web request to respond to the client, but your current handler cannot respond to the client until the request is completed (even if it was async)

Consider separating the request logic off into another process, so it can happen at its own speed.

The key being that you can put work into a queue of some kind to complete eventually, without directly needing the result for response to the client.

You could use an async http request library and some collection of callbacks, multiprocessing to spawn a new process(es), or something more exotic like an independent program (perhaps with a pipe or sockets to communicate).

Maybe something of this form will work for you

import base64
import json
import multiprocessing

URL_EXTERNAL_SERVICE = "https://example.com"
TIMEOUT_REQUESTS = (2, 10)  # always set a timeout for requests
SHARED_QUEUE = multiprocessing.Queue()  # may leak as unbounded

async def slow_request(data):
    SHARED_QUEUE.put(data)
    # now returns on successful queue put, rather than request completion

def requesting_loop(logger, Q, url, token):
    while True:  # expects to be a daemon
        data = json.dumps(Q.get())  # block until retrieval (non-daemon can use sentinel here)
        response = requests.post(
            url,
            json=data,
            headers={'Auth-Header': token},
            timeout=TIMEOUT_REQUESTS,
        )
        # raise_for_status() --> try/except --> log + continue
        if response.status_code != 200:
            logger.error('call to {} failed (code={}) with data: {}'.format(
                url, response.status_code,
                "base64:" + base64.b64encode(data.encode())
            ))

def startup():  # run me when starting
    # do whatever is needed for logger
    # create a pool instead if you may need to process a lot of requests
    p = multiprocessing.Process(
        target=requesting_loop,
        kwargs={"logger": logger, "Q": SHARED_QUEUE, "url": URL_EXTERNAL_SERVICE, "token": settings.API_TOKEN},
        daemon=True
    )
    p.start()
user_45410450
Related