خب خب خب، ‏Middleware های FastAPI🚀
خب middleware یه کد واسط بین دریافت درخواست و پاسخ دادن توی یه اپلیکیشن وبه. یعنی هر ریکوئستی که به سرور میرسه، قبل از رسیدن به route اصلی، از middleware رد میشه و همچنین هر درخواستی هم قبل از رسیدن به کلاینت از middleware عبور میکنه تا تغییر داده بشه، لاگ بشه و ...

چطور توی FastAPI ازشون استفاده کنیم؟🤔
‌‏Middleware ها توی FastAPI با دکوریتور app.middleware تعریف میشن و معمولا ساختارشون این شکلیه:

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def my_middleware(request: Request, call_next):
    # Before reaching route
    print("Before route")
    
    # Executing View
    response = await call_next(request)

    # After view did its thing
    print("After route")

    return response

همونطور که توی مثال بالا دیدید، هر کدی که قبل از اجرای call_next باشه مربوط به درخواست، و هر کدی که بعد از اجرای call_next نوشته بشه مربوط به پاسخ میشه.

توی این مثال قبل از رسیدن درخواست به route اصلی، عبارت 'Before route' چاپ میشه و بعد اینکه route پردازشش با درخواست تموم شد و پاسخ آماده ی برگشت به کلاینت بود، عبارت 'After route' چاپ میشه و بعد از اون پاسخ به کلاینت میرسه.

چندتا مثال ساده برای درک کاربرداش🛠️
از چندتا مثال ساده میتونیم استفاده کنیم
لاگ گیری ساده درخواست ها:

@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"New request: {request.iss.onethod} {request.url}")
    response = await call_next(request)
    return response

اضافه کردن Header به پاسخ:

@app.middleware("http")
async def add_custom_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-App-Version"] = "1.0.0"
    return response

سنجش مدت زمان اجرای درخواست:

import time

@app.middleware("http")
async def measure_time(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    response.headers["X-Process-Time"] = str(duration)
    return response

چندتا نکته📎
1️⃣کلاس Middleware که از FastAPI ایمپورت شده درواقع همون کلاس Middleware توی Starlette هست و برای راحتی میتونیم از FastAPI ایمپورتش کنیم.

2️⃣اگه dependency ای داشته باشیم که با yield تعریف شدن، بخش خرجی اون ها بعد از اجرای middleware اجرا میشه.

3️⃣اگه BackgrounTask ای وجود داشته باشه، اون ها بعد از اجرای همه ی middleware ها اجرا میشن.

جمع بندی✍️
با استفاده از middleware ها میتونید درخواست و پاسخ هارو تمیز کنید، یا قبل از پردازش شدنشون توسط route کارایی که میخواین رو انجام بدین. درواقع با تعریف middleware یه تابع بین client و route اصلی قرار میدیم.

#️⃣ #fastapi #python #backend

➖➖➖➖➖➖➖➖➖➖

🥷🏻 CHANNEL | GROUP

خب خب خب، تسک های CPU Bound🔧
احتمالا CPU Bound به گوشتون خورده، همون تسک هایی که بار پردازشی سنگینی دارن. یعنی دیگه منتظر عملیات I/O(خوندن از فایل یا دیتابیس، API خارجی و...) نیستن و عملا انجام شدنشون توسط CPU طول میکشه.

‏CPU Bound🧮
خب وقتی که به صورت عادی برناممون رو توسعه میدیم، برنامه ی ما فقط توی یک Process و یک هسته ی CPU اجرا میشه. اگه برنامه کار سنگینی مثل پردازش تصویر، محاسبات ریاضی فوق سنگین، الگوریتم های رمزنگاری و فشرده سازی و... داشته باشه، نتیجه اجرا شدنش توی یک Process چیزی بجز کندی و فشار روی CPU نیست. برای مثال شما یه CPU با ۸ هسته دارید ولی برنامه ای که نوشتید فقط روی یک هسته اجرا میشه.
توی این موقعیت میتونیم از Multi Processing استفاده کنیم.

‏Multi Processing یعنی چی؟🧐
میتونیم با استفاده از Multi Processing برنامه ای که نوشتیم رو توی چند Process و روی چند هسته اجرا کنیم. اینجوری از تمام توان CPU استفاده میشه و در نتیجه هم برنامه ما سریع تر میشه و هم فشار روی هسته های CPU تقسیم میشه.

چرا Multi Threading مناسب نیست؟💈
توی پایتون برای تسک های CPU ‌‌‌Bound نمیتونیم از Multi Threading استفاده کنیم. چون پایتون یه چیزی به اسم GIL داره که باعث میشه فقط یک Thread بتونه در لحظه اجرا بشه.
یعنی اگه برای تسک های CPU Bound ازش استفاده کنیم در عمل فقط یک Thread داره واقعا اجرا میشه.

استفاده از Multi Processing🛠
با استفاده از ماژول multiprocessing میتونیم از محدودیت GIL عبور کنیم و چند process داشته باشیم.
یه مثال ساده:

from multiprocessing import Process

def cpu_bound_task():
    # مثلاً محاسبه‌ی یک عدد بزرگ
    total = 0
    for i in range(10**7):
        total += i
    print(total)

if __name__ == '__main__':
    processes = []
    for _ in range(4):
        p = Process(target=cpu_bound_task)
        p.start()
        processes.append(p)

    for p in processes:
        p.join()

این کد تابع cpu_bound_task رو همزمان توی ۴ تا process اجرا میکنه. هر process حافظه، thread و هسته ی CPU خودش رو داره. با اینکار میتونیم واقعا تسک های CPU Bound رو موازی اجرا کنیم و از مزایایی مثل سرعت بهتر بهره مند شیم.

اگه پروژتون بزرگه بهتره تسک های سنگین رو به سیستم هایی مثل Celery بسپرین و از worker های process-based استفاده کنید.

جمع بندی✍️
در نهایت، وقتی با برنامه‌ای سروکار داریم که CPU-bound هست، خیلی مهمه که درست تشخیص بدیم چه راه‌حلی برای بهینه‌سازی استفاده از منابع لازم داریم. توی پایتون، وقتی از multi threading استفاده می‌کنیم، به دلیل محدودیت GIL، همه‌ی پردازش‌ها روی یه هسته و یه thread اجرا می‌شن.
برای این‌که بتونیم از چند هسته‌ی CPU استفاده کنیم و پردازش‌های سنگین رو سریع‌تر انجام بدیم، باید از multi processing بهره بگیریم. با این روش، می‌تونیم هر بخش از برنامه رو به یک process جداگانه اختصاص بدیم که به طور مستقل و هم‌زمان روی هسته‌های مختلف CPU اجرا بشه.

#️⃣ #programmin #python

➖➖➖➖➖➖➖➖➖➖

🥷🏻 CHANNEL | GROUP

خب خب خب، ‏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

خب خب خب، مدیریت تنظیمات با Pydantic⚙️
احتمالا اسم Pydantic به گوشتون خورده یا توی پروژه هاتون ازش استفاده کردین. ولی برای مدیریت تنظیمات پروژه از چی استفاده کردین؟ Pydantic یه کلاس به اسم BaseSettings ارائه میده که برای مدیریت تنظیمات برنامه از جمله متغیر های محیطی و پیش فرض ها استفاده میشه.

چرا از BaseSettings استفاده کنیم؟🧐
1️⃣ خواندن خودکار متغیر های محیطی:
با استفاده از BaseSettigns بدون نیاز به کتابخونه های اضافی مثل python-dotenv میتونیم به متغیر های محیطی دسترسی داشته باشیم.

2️⃣ اعتبارسنجی امن و خودکار متغیر ها:
‏Pydantic به صورت خودکار تایپ هارو چک میکنه و دیگه نیازی به type cast دستی نیست.

3️⃣ پشتیبانی ساده از چند محیط:
با یکم خلاقیت می‌تونیم چندین کلاس تنظیمات برای محیط‌های مختلف مثل Dev, Prod و Test بسازیم.

4️⃣ مناسب برای پروژه‌های بزرگ:
میشه همه‌ی تنظیمات پروژه مثل دیتابیس، کلیدهای API، حالت دیباگ و... رو توی یک کلاس متمرکز نگهداری کرد و راحت تو کل پروژه استفاده‌شون کرد.

ساختار پایه ی کلاس تنظیمات🔧
قبل از هر کاری مطمئن بشید که pydantic و پکیج pydantic-settings نصب باشن:

pip install pydantic pydantic-settings

و بعد میتونید با تعریف کلاس و تنظیماتی که نیاز دارین اونارو مدیریت کنین.

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file='.env', # env file location
    )
    debug: bool = True
    database_url: str
    secret_key: str

این یعنی:
مقدار debug به طور پیش فرض True هست و بقیه مقادیر باید توی فایل env. یا محیط سیستم تعریف بشن، در غیر این صورت ارور میگیریم. نکته جالب اینه که Pydantic از type cast خودکار پشتیبانی میکنه، مثلا اینجا مقدار "True" توی محیط رو به bool تبدیل میکنه.
و فایل env. باید به این شکل باشه:

DATABASE_URL=postgresql://user:pass@localhost/dbname
SECRET_KEY=s3cr3t-k3y

نحوه ی استفاده از تنظیمات🛠
میتونیم یه نمونه از کلاس تنظیمات بسازیم و بعد با استفاده از اتریبیوت ها به تنظیمات دسترسی پیدا کنیم:

settings = Settings()
print(settings.database_url)
print(settings.debug)

جمع بندی✍️
کلاس BaseSettings یکی از ابزارهای بسیار مهم و کاربردی در Pydantic هست که به شما اجازه می‌ده تنظیمات پروژه رو به شکل متمرکز، امن، قابل تست و قابل توسعه مدیریت کنین.
توی پروژه‌های بزرگ یا اپلیکیشن‌هایی که چند محیط (مثل dev، test و prod) دارن، استفاده از این ساختار کمک می‌کنه کد تمیزتر و حرفه‌ای‌تری داشته باشین.

#️⃣ #programming #python

➖➖➖➖➖➖➖➖➖➖

🥷🏻 CHANNEL | GROUP