0Pricing
FastAPI Backend Development Bootcamp · درس

تفريغ المهام الخفيف باستخدام BackgroundTasks

استخدام BackgroundTasks المدمجة في FastAPI لتنفيذ الآثار الجانبية بنمط fire-and-forget دون حجب الاستجابة.

تفريغ المهام الخفيف باستخدام BackgroundTasks درس مجاني في FastAPI Backend Development Bootcamp على CoddyKit. هذا هو الدرس 1 من أصل 4. يمكنك قراءة الدرس كاملاً أدناه مجاناً — ثم تمرن عليه مباشرة في المتصفح باستخدام محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7. هذا الدرس جزء من مسار التعلم في FastAPI Backend Development Bootcamp، وتقدمك يتزامن عبر الويب وتطبيق CoddyKit. تتضمن دورة FastAPI Backend Development Bootcamp 4 دروس في المجموع.

بعض أجزاء هذا الدرس لم تُترجم بعد وتظهر باللغة الإنجليزية.

Why Offload Work?

When a client sends a request, they wait for the response. If your endpoint also sends a welcome email, writes an audit log, or warms a cache, the user is stuck waiting for work they don't care about.

Fire-and-forget side effects are tasks that should run after the response is sent, without blocking it:

  • Sending notification emails
  • Writing analytics or audit logs
  • Invalidating or warming caches
  • Cleaning up temporary files

FastAPI ships a built-in tool for exactly this: BackgroundTasks.

Declaring BackgroundTasks

To use it, add a parameter typed as BackgroundTasks to your path operation function. FastAPI sees the type annotation and injects an instance for you, just like any other dependency.

You then register work with .add_task(func, *args, **kwargs). The function is not called immediately, it is queued to run once the response has been returned.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_log(message: str) -> None:
    with open("log.txt", mode="a") as f:
        f.write(message + "\n")


@app.post("/signup")
async def signup(email: str, tasks: BackgroundTasks):
    tasks.add_task(write_log, f"signup: {email}")
    return {"status": "accepted"}

The Execution Order

The critical detail: background tasks run after the response is sent to the client, but still within the same server process.

  • The endpoint returns its dict or Response.
  • FastAPI flushes the response over the network.
  • Only then does it execute each queued task, in the order they were added.

So the user gets an instant 202-style reply while the email or log happens behind the scenes.

Passing Arguments to a Task

Arguments you pass to add_task are stored and forwarded when the task finally runs. Positional and keyword arguments both work.

This pattern keeps the side-effect logic in a plain function that is easy to unit-test in isolation, completely independent of FastAPI.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def send_email(to: str, subject: str, body: str) -> None:
    # imagine an SMTP client here
    print(f"Sending to {to}: {subject}")


@app.post("/orders")
async def create_order(email: str, tasks: BackgroundTasks):
    order_id = 1234
    tasks.add_task(
        send_email,
        to=email,
        subject="Order confirmed",
        body=f"Your order {order_id} is on the way!",
    )
    return {"order_id": order_id}

Sync vs Async Task Functions

A task function can be either a normal def or an async def.

  • An async task is awaited directly on the event loop.
  • A regular def task is run in a thread pool so it doesn't block the loop.

Rule of thumb: if your side effect does blocking I/O (file writes, a synchronous DB driver), a plain def is fine, FastAPI offloads it to a thread. Use async def only when you genuinely await async I/O.

async def notify_async(user_id: int) -> None:
    # awaits an async HTTP client, for example
    await some_async_push(user_id)


def notify_sync(user_id: int) -> None:
    # blocking call, run in a threadpool by FastAPI
    requests_post(user_id)

Adding Multiple Tasks

You can call add_task as many times as you like. Tasks run sequentially in the exact order added, each completing before the next begins.

Because they run one after another, a slow task delays the ones queued behind it, but never the HTTP response itself.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


@app.post("/publish")
async def publish(post_id: int, tasks: BackgroundTasks):
    tasks.add_task(reindex_search, post_id)
    tasks.add_task(invalidate_cache, post_id)
    tasks.add_task(notify_followers, post_id)
    return {"published": post_id}

Using BackgroundTasks in Dependencies

A powerful trick: a dependency can also declare a BackgroundTasks parameter and queue tasks. FastAPI merges everything into one shared task set for that request.

This lets cross-cutting concerns, like audit logging, live in a reusable dependency instead of being copy-pasted into every endpoint.

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def audit(action: str, tasks: BackgroundTasks):
    tasks.add_task(write_audit_row, action)
    return action


@app.delete("/items/{item_id}")
async def delete_item(item_id: int, action=Depends(audit)):
    return {"deleted": item_id}

A Plain-Python Task Queue Mental Model

Under the hood, BackgroundTasks is little more than a list of callables that get run after the response. You can model the idea in pure Python to build intuition.

The snippet below is standalone, no FastAPI needed, showing the add-then-run-later pattern.

class TaskList:
    def __init__(self):
        self.tasks = []

    def add_task(self, func, *args, **kwargs):
        self.tasks.append((func, args, kwargs))

    def run_all(self):
        for func, args, kwargs in self.tasks:
            func(*args, **kwargs)


def log(msg):
    print("LOG:", msg)


q = TaskList()
q.add_task(log, "user signed up")
q.add_task(log, "email queued")
print("response sent")
q.run_all()

Error Handling Inside Tasks

Because a task runs after the response, you can no longer turn its failure into an HTTP error, the client already got a 200.

An unhandled exception in a background task is logged by the server but is invisible to the client. Always wrap risky work in try/except and decide on retries or a dead-letter strategy yourself.

def send_receipt(order_id: int) -> None:
    try:
        deliver_email(order_id)
    except Exception as exc:
        # the client already has its 200, so log and recover here
        logger.exception("receipt failed for %s: %s", order_id, exc)
        schedule_retry(order_id)

The Big Limitation: Same Process

BackgroundTasks runs in the same worker process as your app. That brings real constraints:

  • Heavy CPU work still consumes that worker's resources.
  • If the process crashes or is redeployed, queued tasks are lost, there is no persistence.
  • Tasks don't survive across multiple machines or scale horizontally.

It is perfect for lightweight, best-effort side effects, but not for reliable, long-running, or distributed jobs.

When to Reach for Celery Instead

Choose BackgroundTasks when the work is short, non-critical, and OK to lose on a crash, sending an email, bumping a counter, deleting a temp file.

Reach for Celery or another distributed queue (RQ, Dramatiq, Arq) when you need:

  • Durability, jobs survive restarts via a broker like Redis/RabbitMQ.
  • Retries, scheduling, and rate limiting.
  • Horizontal scaling across dedicated worker machines.
  • Heavy CPU jobs that would otherwise starve your web workers.

Quick Check

Test your understanding of when BackgroundTasks is the right tool.

Recap

Key takeaways:

  • Add a BackgroundTasks parameter and call add_task(func, *args, **kwargs) to defer side effects.
  • Tasks run after the response, sequentially, in the same worker process.
  • Sync def tasks run in a thread pool; async def tasks run on the event loop.
  • Dependencies can queue tasks too, great for cross-cutting concerns like auditing.
  • No persistence: failures are invisible to the client and tasks die with the process.
  • Use it for lightweight, best-effort work; choose Celery for durable, retryable, distributed, or CPU-heavy jobs.

الأسئلة الشائعة

هل درس «تفريغ المهام الخفيف باستخدام BackgroundTasks» مجاني؟

نعم — نص درس «تفريغ المهام الخفيف باستخدام BackgroundTasks» كامل متاح مجاناً هنا على الويب. لتمرينه بشكل تفاعلي (محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7) وفتح باقي دورة FastAPI Backend Development Bootcamp، انتقل إلى CoddyKit PRO. تتضمن دورة FastAPI Backend Development Bootcamp 4 دروس في المجموع.

ماذا ستتعلم في «تفريغ المهام الخفيف باستخدام BackgroundTasks»؟

استخدام BackgroundTasks المدمجة في FastAPI لتنفيذ الآثار الجانبية بنمط fire-and-forget دون حجب الاستجابة. تتمرن على FastAPI Backend Development Bootcamp مع أكواد عملية تشغلها مباشرة في المتصفح، ومدرس ذكاء اصطناعي متاح 24/7 يجيب على أسئلتك أثناء عملك.

هل أحتاج إلى خبرة سابقة لأبدأ FastAPI Backend Development Bootcamp؟

لا تُشترط خبرة سابقة. FastAPI Backend Development Bootcamp على CoddyKit منظم للمبتدئين حتى المتقدمين، لذا يمكنك البدء من هنا أو من البداية والتقدم بسرعتك الخاصة. هذا هو الدرس 1 من أصل 4.

كم من الوقت يستغرق درس «تفريغ المهام الخفيف باستخدام BackgroundTasks»؟

معظم دروس CoddyKit تستغرق حوالي 5–10 دقائق. كل منها موجز وتفاعلي، لذا تحرز تقدماً مستمراً وتستأنف من حيث توقفت عبر الويب والتطبيق.

هل يمكنني كتابة وتشغيل أكواد في درس FastAPI Backend Development Bootcamp هذا؟

نعم. كل درس في FastAPI Backend Development Bootcamp يتضمن محرر أكواد مدمج، لذا تكتب وتشغل أكواداً حقيقية مباشرة في متصفحك وتحصل على تعليقات فورية من الذكاء الاصطناعي — بدون إعداد محلي.

جميع الدروس في هذه الدورة

  1. تفريغ المهام الخفيف باستخدام BackgroundTasks
  2. ربط Celery Workers بتطبيق FastAPI
  3. إعادة المحاولة وقابلية التكرار ومعالجة الرسائل الميتة
  4. المهام المجدولة والدورية باستخدام Celery Beat
← العودة إلى FastAPI Backend Development Bootcamp