Ninja Learn | نینجا لرن

خب خب خب، ‏Background Task ها توی FastAPI🚀
گاهی اوقات نیاز داریم که یه کاری بعد از ارسال پاسخ به کاربر انجام بشه. مثل ارسال ایمیل خوشامد گویی، ثبت لاگ یا آمار توی دیتابیس، پردازش فایل آپلود شده و .... توی این شرایط میتونیم از Background Task ها استفاده کنیم، اینجوری میتونیم بدون معطل کردن کاربر اون کارهارو جداگانه انجام بدیم.

استفاده از Background Task ها🛠
خب اول باید کلاس BackgroundTasks رو ایمپورت کنیم و یه پارامتر از همین نوع برای فانکشن route بنویسیم.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

حالا FastAPI میاد یه آبجکت با نوع BackgroundTasks برامون ایجاد میکنه و به اون پارامتر پاس میده.
بعد از اینکه فانکشن تسکمون رو ایجاد کردیم میتونیم با استفاده از متود ()add_task از همون پارامتر اون فانکشن رو به صف اجرا اضافه کنیم. همچنین میتونیم آرگومان های مورد نیازمون رو هم با استفاده از همین متود به تسکمون پاس بدیم.

‏Background Tasks و Dependency injection💉
‏Background Tasks به خوبی با سیستم تزریق وابستگی FastAPI سازگاره. میتونیم توی سطح های مختلف برنامه(فانکشن route، یه وابستگی و...) از Background Task استفاده کنیم.

from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

توی این مثال بعد از اینکه پاسخ به کلاینت ارسال شد، یه تسک میاد ایمیل کاربر رو توی فایل log.txt مینویسه. اگه یه کوئری پارامتر هم به API ارسال بشه یه تسک دیگه اون رو هم توی فایل مینویسه.

نکته مهم⚠️
این ابزار فقط برای کارهای سبک وسریع مناسبه. مثل همین لاگ نوشتن، ارسال ایمیل یا پردازش های خیلی کوچیک و سبک. برای کارهای سنگین تر مثل پردازش تصویر بهتره که از سیستم هایی مثل Celery استفاده بشه.

جمع بندی✍️
‏Background Task یه ابزار ساده ولی کاربردیه. میتونه توی پروژه هایی که تسک های سنگینی ندارن از Celery بی نیازتون کنه و کارهارو بعد از پاسخ دهی به صورت غیرهمزمان انجام بده.

#️⃣ #fastapi #python #backend

➖➖➖➖➖➖➖➖➖➖

🥷🏻 CHANNEL | GROUP

👍11❤6

www.tgoop.com/ninja_learn_ir/779

1.9K viewsDenver, May 9 at 10:53

خب خب خب، ‏Background Task ها توی FastAPI🚀
گاهی اوقات نیاز داریم که یه کاری بعد از ارسال پاسخ به کاربر انجام بشه. مثل ارسال ایمیل خوشامد گویی، ثبت لاگ یا آمار توی دیتابیس، پردازش فایل آپلود شده و .... توی این شرایط میتونیم از Background Task ها استفاده کنیم، اینجوری میتونیم بدون معطل کردن کاربر اون کارهارو جداگانه انجام بدیم.

استفاده از Background Task ها🛠
خب اول باید کلاس BackgroundTasks رو ایمپورت کنیم و یه پارامتر از همین نوع برای فانکشن route بنویسیم.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

#️⃣ #fastapi #python #backend

🥷🏻 CHANNEL | GROUP

BY Ninja Learn | نینجا لرن