5. FastAPI Async Await Hindi – Non-Blocking Code 2026

नमस्ते दोस्तों!

स्वागत है The Easy Master पर!

क्या आपका FastAPI server slow lagta hai? जब एक user का request process हो रहा हो, तो दूसरे users को wait karna padta hai? ऐसा लगता है jaise ek queue लगी हो?

ये problem तब आती है jab aapka code blocking होता है। Matlab – एक request पूरी होने तक दूसरी start नहीं होती।

मेरे साथ भी ऐसा हुआ। मैंने एक API बनाई जो database query करती थी – वो query 2 seconds लेती थी। जब 10 users ने साथ में call kiya, toh sabko 20 seconds lag gaye! क्योंकि code blocking था।

तब मैंने async/await सीखा – और server फिर से fast ho gaya.

इस FastAPI async await Hindi article में मैं आपको सब कुछ सिखाऊंगा:

✅ Async/await kya hai – गोलू-मोलू example ke saath
✅ Synchronous (blocking) vs Asynchronous (non-blocking)
✅ FastAPI में async def कब use karein, कब def
✅ CPU-bound vs I/O-bound tasks
✅ Async database calls – SQLAlchemy + databases
✅ await का सही use – बिना await क्या होगा?
✅ Background tasks aur asyncio.create_task
✅ Common mistakes – blocking code inside async function
✅ Real project example – async file upload + email send

End mein cheat sheet और practice prompts bhi.

चलिए शुरू करते हैं! 🚀

1. Async/Await Kya Hai? – Simple Story

मान लो तुम एक restaurant में chef हो।

Synchronous (blocking) style:
तुम एक order लेते हो – उसे बनाने लगते हो – जब तक वो order पूरा नहीं होता, दूसरा order accept नहीं कर सकते। बीच में customer wait करता है। ये inefficient है।

Asynchronous (non-blocking) style:
तुम order लेते हो – उसे stove पर रख देते हो (जो अपने आप पकता है) – और दूसरे order ले लेते हो। जब पहला order ready होता है, तुम उसे serve कर देते हो। एक साथ multiple orders handle हो रहे हैं।

FastAPI में async/await यही करता है – जब कोई task I/O (database, network, file) पर wait कर रहा है, तो CPU दूसरे requests handle कर सकता है।

2. Synchronous vs Asynchronous – Difference

Synchronous (Normal Python function) – Blocking

import time

@app.get("/sync-example")
def sync_endpoint():
    time.sleep(3)  # Blocking call – CPU yahi rukega
    return {"message": "Done after 3 seconds"}

अगर 10 requests आएंगी, तो हर एक 3 seconds legi – total ~30 seconds.

Asynchronous (async def) – Non-blocking

import asyncio

@app.get("/async-example")
async def async_endpoint():
    await asyncio.sleep(3)  # Non-blocking – CPU dusre kaam kar sakta hai
    return {"message": "Done after 3 seconds"}

10 requests – लगभग 3 seconds में सब finish हो जाएंगी (parallelism not exactly, but concurrency).

Visual difference:

3. FastAPI Mein Async Def vs Def

FastAPI दोनों support करता है। कब क्या use karein?

Use async def when:

• I/O operations (database, API calls, file read/write)
• Multiple await calls inside function
• Need concurrenct

import aiohttp

@app.get("/external-api")
async def call_external():
    async with aiohttp.ClientSession() as session:
        response = await session.get("https://api.example.com/data")
        return await response.json()

Use normal def when:

• Only CPU-bound computation (math, data processing)
• Simple return without any await
• Using blocking libraries (like requests, standard open)

@app.get("/compute")
def compute_square(x: int):
    return {"result": x * x}  # no I/O, no need async

Important: अगर async def के अंदर blocking code डालोगे (e.g., time.sleep, requests.get), तो performance और खराब हो जाएगी। क्योंकि async function event loop को block कर देगी।

4. I/O-Bound vs CPU-Bound Tasks

I/O-Bound Tasks (Async ke liye perfect)

जहाँ program network, disk, database पर wait करता है।

Examples:

• Database queries
• HTTP requests (to another API)
• File read/write
• Sending emails
• Sleeping (asyncio.sleep)

Solution: Use async/await – multiple tasks can run concurrently.

CPU-Bound Tasks (Async help nahi karega)

जहाँ processor lagatar computation kar raha hai.

Examples:

• Image processing
• Machine learning inference
• Complex mathematical calculations
• Video encoding

Solution: Use def normal function + background worker (Celery, etc.), ya asyncio.to_thread use karo.

import asyncio

def heavy_computation(n):
    # CPU-bound task
    return sum(i*i for i in range(n))

@app.get("/heavy")
async def heavy_endpoint(n: int):
    # Run blocking CPU task in separate thread
    result = await asyncio.to_thread(heavy_computation, n)
    return {"result": result}

5. Async Database Calls – Real Example

Synchronous database drivers (like psycopg2, sqlite3) blocking होते हैं। Async ke liye async drivers use karo:

• PostgreSQL: asyncpg + databases or SQLAlchemy async
• MySQL: aiomysql
• MongoDB: motor

Example using databases library:

pip install databases[postgresql] asyncpg

from fastapi import FastAPI
from databases import Database
import asyncio

app = FastAPI()
database = Database("postgresql://user:pass@localhost/db")

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Non-blocking database query
    query = "SELECT * FROM users WHERE id = :id"
    user = await database.fetch_one(query=query, values={"id": user_id})
    return user

Async SQLAlchemy example:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession)

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(Item).where(Item.id == item_id))
        return result.scalar_one()

6. Multiple Concurrent Requests – Performance Test

Code to test difference:

import asyncio
import time
from fastapi import FastAPI

app = FastAPI()

@app.get("/sync")
def sync_endpoint():
    time.sleep(2)  # blocking
    return {"status": "ok"}

@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(2)  # non-blocking
    return {"status": "ok"}

Test using httpx or ab (Apache Bench):

# Install httpx
pip install httpx

# Python test script
import asyncio
import httpx
import time

async def test_concurrent(url, n=10):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for _ in range(n)]
        start = time.time()
        await asyncio.gather(*tasks)
        end = time.time()
        print(f"{n} requests took {end-start:.2f} seconds")

# Run
asyncio.run(test_concurrent("http://localhost:8000/sync", 10))
asyncio.run(test_concurrent("http://localhost:8000/async", 10))

Result:

• Sync: 10 requests → ~20 seconds
• Async: 10 requests → ~2 seconds

Personal Experience: मैंने अपने एक project में async database calls implement किए, तो response time 500ms से घटकर 50ms हो गया – क्योंकि concurrent requests parallel mein process हो रही थीं।

7. Await Ka Magic – Bina await Bhoolna

await keyword batata hai ki FastAPI wahan wait karega, beech mein dusre requests handle karega.

Galti #1 – await bhoolna:

@app.get("/bad")
async def bad_example():
    # Yeh galat hai – coroutine ko await karna padega
    asyncio.sleep(2)  # ❌ missing await
    return {"message": "hello"}

FastAPI warning dega: RuntimeWarning: coroutine 'sleep' was never awaited

Sahi tarika:

@app.get("/good")
async def good_example():
    await asyncio.sleep(2)  # ✅ correct
    return {"message": "hello"}

Galti #2 – async function ke andar blocking library:

import requests  # blocking library

@app.get("/wrong")
async def wrong():
    response = requests.get("https://api.example.com")  # ❌ blocks event loop
    return response.json()

Sahi tarika:

import aiohttp  # async HTTP client

@app.get("/right")
async def right():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as resp:
            return await resp.json()

8. Background Tasks – Asyncio.Create_Task

Kabhi kabhi aapko request ka response immediately dena hai, par background mein kaam karna hai – jaise email send karna, log save karna.

Option 1: FastAPI BackgroundTasks

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # Blocking code chalega background thread mein
    time.sleep(5)  # simulate sending
    print(f"Email sent to {email}")

@app.post("/register/")
async def register_user(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"message": "User registered, email will be sent"}

Option 2: asyncio.create_task (for async background tasks)

import asyncio

async def send_async_email(email: str):
    await asyncio.sleep(3)
    print(f"Async email sent to {email}")

@app.post("/register-async")
async def register_async(email: str):
    # Fire and forget – no await
    asyncio.create_task(send_async_email(email))
    return {"message": "Registered"}

Caution: create_task ke tasks survive nahi karte agar application shutdown ho jaye. Production ke liye use Celery or RQ.

9. Real Project – File Upload + Email Notification

import asyncio
import aiofiles
from fastapi import FastAPI, UploadFile, File, BackgroundTasks
from fastapi.concurrency import run_in_threadpool
import smtplib  # blocking, but we'll use threadpool

app = FastAPI()

# Async file upload
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    # Write file asynchronously
    async with aiofiles.open(f"uploads/{file.filename}", "wb") as f:
        content = await file.read()
        await f.write(content)
    # Return immediately, email in background
    asyncio.create_task(send_email_notification(file.filename))
    return {"filename": file.filename, "status": "uploaded"}

async def send_email_notification(filename: str):
    # Simulate email send (non-blocking)
    await asyncio.sleep(1)
    print(f"Email: File {filename} uploaded")
    # Actually use aiosmtplib for async email

# CPU-heavy processing – using threadpool
async def process_image(image_path: str):
    # CPU bound – run in thread
    def process():
        # heavy image processing
        return "processed"
    result = await run_in_threadpool(process)
    return result

@app.post("/process-image/")
async def process_image_endpoint(file: UploadFile):
    await file.save(f"temp/{file.filename}")
    result = await process_image(f"temp/{file.filename}")
    return {"result": result}

10. Common Mistakes (aur Unka Solution!)

Personal Mistake: मैंने एक बार async function के अंदर pandas का heavy operation run kar diya – server hang ho gaya. फिर pata chala ki CPU-bound tasks async event loop ko block karte hain. तब से CPU tasks के लिए asyncio.to_thread use karta hoon.

11. Best Practices – Professional Code

✅ I/O-bound tasks → async def + await async libraries
✅ CPU-bound tasks → def या asyncio.to_thread
✅ Database → Use async drivers (asyncpg, aiomysql, motor)
✅ HTTP calls → httpx.AsyncClient (recommended) or aiohttp
✅ File I/O → aiofiles for async file operations
✅ Never use time.sleep in async functions – use asyncio.sleep
✅ Use BackgroundTasks for simple background work
✅ Use asyncio.gather for parallel async tasks:

async def parallel_example():
    results = await asyncio.gather(
        task1(),
        task2(),
        task3()
    )
    return results

✅ Set timeouts for async operations:

try:
    result = await asyncio.wait_for(long_task(), timeout=5.0)
except asyncio.TimeoutError:
    return {"error": "Timeout"}

12. Resources – Cheat Sheet + Practice Prompts

Cheat Sheet (Copy-Paste)

# Basic async endpoint
@app.get("/")
async def hello():
    await asyncio.sleep(0.1)
    return {"msg": "Hello"}

# Parallel execution
@app.get("/parallel")
async def parallel():
    a, b = await asyncio.gather(task1(), task2())
    return {"a": a, "b": b}

# Run blocking code in thread
result = await asyncio.to_thread(blocking_func, arg)

# Async HTTP client (httpx)
async with httpx.AsyncClient() as client:
    resp = await client.get("https://api.example.com")

# Async file write
async with aiofiles.open("file.txt", "w") as f:
    await f.write("data")

# Background task
bg.add_task(send_email, email)

# Timeout
result = await asyncio.wait_for(coro(), timeout=10)

Practice Prompts

1. Beginner: Ek async endpoint banao jo asyncio.sleep(2) kare aur current timestamp return kare. Do concurrent requests test karo.
2. Intermediate: Ek async endpoint banao jo 3 alag-alag external APIs ko call kare (e.g., JSONPlaceholder) using asyncio.gather. Saare responses ko combine karke return karo.
3. Advanced: Ek file upload endpoint banao jo file read kare, CPU-heavy processing (e.g., image resizing) background mein threadpool mein kare, aur result save kare – while client ko immediate “processing started” response de.

13. FAQ

Q1: Kya FastAPI automatically synchronous functions ko async mein convert kar deta hai?

Nahi. Synchronous functions (def) threadpool mein execute hote hain – but still blocking for that thread. Async functions (async def) event loop pe run hote hain. Choose wisely.

Q2: asyncio.to_thread aur BackgroundTasks mein kya antar hai?

to_thread aapke current request ke andar blocking code ko background thread mein daal deta hai, par response tab tak wait karega jab tak wo complete na ho jaye. BackgroundTasks response ke baad background mein chalata hai – user ko immediately response mil jata hai.

Q3: Kya main async database driver use karu ya sync driver with threadpool?

Production ke liye async driver better hai – less overhead. Lekin agar aapka ORM async support nahi karta (e.g., old SQLAlchemy), toh threadpool acceptable hai.

Q4: FastAPI async routes se multiple requests automatically parallel handle hoti hain?

Yes – लेकिन only I/O wait ke time par. Agar aapka async route CPU-bound hai, toh event loop block ho jayega aur concurrency nahi milegi.

Q5: await asyncio.sleep vs time.sleep – difference kya hai?

asyncio.sleep non-blocking hai – event loop dusre tasks handle kar sakta hai. time.sleep blocking hai – poori thread ruk jati hai.

Q6: Kya main FastAPI mein WebSockets ke saath async use kar sakta hoon?

Haan – FastAPI WebSockets naturally async हैं। @app.websocket("/ws") async def websocket_endpoint(websocket):

Q7: Async routes ke saath debugging mushkil hoti hai?

Thodi – but asyncio debug mode (PYTHONASYNCIODEBUG=1) aur proper logging use karo. Also asyncio.run karne se pehle asyncio.get_running_loop() check karo.

14. Conclusion

बहुत बढ़िया! Aapne aaj seekh liya:

✅ FastAPI async await Hindi – async/await का real meaning
✅ Synchronous vs Asynchronous – kab kya use karna hai
✅ I/O-bound vs CPU-bound tasks
✅ Async database calls, HTTP requests, file operations
✅ Background tasks aur parallel execution (asyncio.gather)
✅ Common mistakes aur best practices

Async/await aapke FastAPI applications ko 10x faster bana sakta hai – बस सही जगह use karo.

Aapki challenge: Upar diye practice prompts mein se “parallel API calls” wala implement karo. Apna code comment mein share karo – main personally review karunga.

Next topic?

• FastAPI WebSockets (Real-time chat app)?
• FastAPI Testing (pytest + async)?
• Deploying FastAPI on production (Gunicorn + Uvicorn)?

Comment mein batao!

The Easy Master ke saath async coding ka maza lo. Non-blocking raho, fast raho! 🚀

Resources

• FastAPI Async Documentation
• Python Asyncio Official Docs
• HTTPX Async Client Guide
• Aiofiles Documentation
• Async SQLAlchemy Tutorial

Additional Resources

• MongoDB Setup – CRUD Operations समझे (Beginner Guide) 2026
• MongoDB Compass से GUI Database Manage करने का आसान तरीका | Beginner Tutorial 2026
• Node.js MongoDB Connection – Mongoose Setup समझे 2026
• MongoDB Data Modeling – Embedded Documents vs References 2026
• MongoDB Indexing – Query Speed बढ़ने करने का आसान तरीका
• MongoDB Aggregation Pipeline – Stages समझे | Practical Examples
• Express MongoDB CRUD – Complete REST API Example
• Mongoose ODM – Schema, Model, and Validation समझे 2026
• MongoDB Atlas – Cloud Database Free में Deploy करें 2026
• JWT Authentication MongoDB – User Login System कैसे बनाएं 2026
• Next.js App Router – Pages Router से क्या बदला? Complete Guide 2026
• Next.js Server Components – ‘use client’ कब और कहां Use करें
• Next.js Routing – Layouts, Dynamic Routes and Nested Routes 2026
• Next.js 16 New Features – Turbopack, Cache Components and proxy.ts
• Next.js Partial Prerendering – Static and Dynamic Content साथ में
• Next.js Authentication – JWT Session Management (App Router) 2026
• Next.js SEO – Metadata API and Image Optimization समझे 2026
• Next.js TypeScript – Full-Stack Type-Safe App कैसे बनाएं
• FastAPI Kya Hai? FastAPI Python Setup Aur Pehla API Hindi 2026
• FastAPI Path Parameters Hindi – शून्य से हीरो तक गाइड 2026
• Pydantic v2 Tutorial Hindi – Data Validation Master 2026
• FastAPI dependency injection Hindi – Code Reuse Ka Magic
  

Views: 282