#8 FastAPI Module 8 — Background Tasks

Module 8 — Background Tasks in FastAPI

FastAPI includes a powerful built-in BackgroundTasks system that lets you run tasks after returning a response — ideal for:

• Sending emails after user signup
• Logging
• Cleanup jobs
• Database audit entries
• Processing heavy operations asynchronously

---

1️⃣ Basic Background Task Example

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def write_log(message: str):
    with open("log.txt", "a") as file:
        file.write(message + "\n")
        time.sleep(2)  # Simulate slow processing

@app.get("/log")
async def log_message(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "User visited /log endpoint")
    return {"message": "Log task scheduled!"}

---

2️⃣ Sending Email in Background

(Fake implementation, no SMTP required)

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, subject: str, body: str):
    with open("emails.txt", "a") as f:
        f.write(f"TO: {to}\\nSUBJECT: {subject}\\nBODY: {body}\\n---\\n")

@app.post("/send-email")
async def trigger_email(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Welcome!", "Thanks for joining our platform.")
    return {"status": "Email sent in background"}

---

3️⃣ Background Tasks with Database Operations

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

fake_db = []

def add_to_db(item: str):
    fake_db.append(item)

@app.post("/add-item")
async def add_item(item: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(add_to_db, item)
    return {"status": "Item scheduled for DB insertion"}

---

4️⃣ Background Tasks + Dependency Injection

from fastapi import Depends, BackgroundTasks, FastAPI

app = FastAPI()

def notify_admin(message: str):
    print("ADMIN ALERT:", message)

def get_admin_notifier():
    return notify_admin

@app.get("/alert")
async def alert_admin(
    background: BackgroundTasks,
    alert=Depends(get_admin_notifier)
):
    background.add_task(alert, "Suspicious activity detected")
    return {"status": "Admin notified in background"}

---

5️⃣ When to Use Background Tasks?

✔ Non-critical operations ✔ Tasks that can safely run “after response” ✔ When Celery / RabbitMQ is not needed ✔ Good for small/medium apps ---