Concurrent task execution in FastAPI - Mini Project: Build & Apply

Concurrent Task Execution with FastAPI

📖 Scenario: You are building a simple FastAPI web service that runs multiple tasks at the same time to save waiting time. This is like asking several friends to do small jobs for you all at once instead of one by one.

🎯 Goal: Create a FastAPI app that runs two tasks concurrently and returns their results together.

📋 What You'll Learn

Create a FastAPI app instance named app

Define two async functions named task_one and task_two that simulate work with asyncio.sleep

Use asyncio.gather to run task_one and task_two concurrently inside a GET endpoint /run-tasks

Return a JSON response with results from both tasks

💡 Why This Matters

🌍 Real World

Running multiple independent tasks at the same time improves speed and responsiveness in web services, like fetching data from different sources simultaneously.

💼 Career

Understanding concurrency in FastAPI is important for backend developers to build efficient APIs that handle multiple operations without delay.

1

Create FastAPI app and async tasks

Import FastAPI and asyncio. Create a FastAPI app instance called app. Define two async functions named task_one and task_two. Each should use await asyncio.sleep(1) and then return the strings 'Task One Complete' and 'Task Two Complete' respectively.

FastAPI

from fastapi import FastAPI
import asyncio

# Your code here

Need a hint?

Remember to import FastAPI and asyncio. Use async def to define asynchronous functions.

2

Add endpoint to run tasks concurrently

Create a GET endpoint /run-tasks using @app.get. Define an async function run_tasks for this endpoint. Inside it, use asyncio.gather to run task_one() and task_two() concurrently and store the results in a variable called results.

FastAPI

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def task_one():
    await asyncio.sleep(1)
    return 'Task One Complete'

async def task_two():
    await asyncio.sleep(1)
    return 'Task Two Complete'

@app.get('/run-tasks')
async def run_tasks():
    # Your code here

Need a hint?

Use @app.get('/run-tasks') to create the endpoint. Use asyncio.gather to run both tasks at the same time.

3

Return results as JSON with keys

Modify the run_tasks function to return a dictionary with keys 'task_one' and 'task_two' mapping to the respective results from results list.

FastAPI

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def task_one():
    await asyncio.sleep(1)
    return 'Task One Complete'

async def task_two():
    await asyncio.sleep(1)
    return 'Task Two Complete'

@app.get('/run-tasks')
async def run_tasks():
    results = await asyncio.gather(task_one(), task_two())
    # Your code here

Need a hint?

Return a dictionary with keys task_one and task_two using the results list.

4

Add docstring and run instructions

Add a docstring to the run_tasks function that says "Run two tasks concurrently and return their results.". Also add a comment at the bottom with the text # Run with: uvicorn filename:app --reload replacing filename with your script name.

FastAPI

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def task_one():
    await asyncio.sleep(1)
    return 'Task One Complete'

async def task_two():
    await asyncio.sleep(1)
    return 'Task Two Complete'

@app.get('/run-tasks')
async def run_tasks():
    # Your code here
    results = await asyncio.gather(task_one(), task_two())
    return {'task_one': results[0], 'task_two': results[1]}

# Add run instructions comment below

Need a hint?

Add a clear docstring inside the run_tasks function. Add a comment at the end with the uvicorn run command.