“`html

Flask و Celery: پیاده‌سازی وظایف پس‌زمینه

در دنیای توسعه وب مدرن، پاسخگویی و کارایی دو رکن اساسی برای ارائه یک تجربه کاربری ایده‌آل هستند. اغلب اوقات، برنامه‌های وب با وظایفی پیچیده و زمان‌بر مواجه می‌شوند که اجرای آن‌ها در حین درخواست کاربر می‌تواند منجر به کندی و ناپاسخگویی شود. در اینجاست که وظایف پس‌زمینه (Background Tasks) به کمک ما می‌آیند. با انتقال این وظایف به یک سیستم پردازش پس‌زمینه، می‌توانیم اطمینان حاصل کنیم که برنامه وب ما سریع و پاسخگو باقی می‌ماند.

در این مقاله جامع، به بررسی نحوه پیاده‌سازی وظایف پس‌زمینه در برنامه‌های Flask با استفاده از Celery خواهیم پرداخت. Celery یک سیستم صف پیام توزیع شده قدرتمند است که به توسعه‌دهندگان اجازه می‌دهد وظایف را به‌صورت غیرهمزمان (Asynchronously) اجرا کنند. این بدان معناست که برنامه Flask ما می‌تواند درخواست‌های کاربر را دریافت کند، وظایف زمان‌بر را به Celery واگذار کند و سپس بلافاصله به کاربر پاسخ دهد. Celery وظایف را در پس‌زمینه پردازش می‌کند و نتایج را در صورت نیاز به برنامه Flask برمی‌گرداند.

چرا از وظایف پس‌زمینه استفاده کنیم؟

استفاده از وظایف پس‌زمینه مزایای متعددی را برای برنامه‌های Flask به ارمغان می‌آورد، از جمله:

• بهبود پاسخگویی: با انتقال وظایف زمان‌بر به پس‌زمینه، برنامه Flask ما می‌تواند سریع‌تر به درخواست‌های کاربر پاسخ دهد و تجربه کاربری بهتری را ارائه کند.
• مقیاس‌پذیری: Celery به ما امکان می‌دهد وظایف را در چندین ماشین یا فرآیند توزیع کنیم، که این امر مقیاس‌پذیری برنامه ما را به میزان قابل توجهی افزایش می‌دهد.
• انعطاف‌پذیری: Celery از انواع مختلفی از کارگزاران پیام (Message Broker) مانند RabbitMQ و Redis پشتیبانی می‌کند، که به ما انعطاف‌پذیری بیشتری در انتخاب زیرساخت مناسب برای برنامه خود می‌دهد.
• قابلیت اطمینان: Celery دارای مکانیسم‌های داخلی برای مدیریت خطاها و تلاش مجدد وظایف است، که این امر قابلیت اطمینان برنامه ما را افزایش می‌دهد.

برخی از کاربردهای رایج وظایف پس‌زمینه عبارتند از:

• ارسال ایمیل
• پردازش تصویر و ویدئو
• محاسبات پیچیده
• همگام‌سازی داده‌ها
• وظایف زمان‌بندی شده

پیش‌نیازها

قبل از شروع پیاده‌سازی وظایف پس‌زمینه با Flask و Celery، باید اطمینان حاصل کنیم که پیش‌نیازهای زیر برآورده شده‌اند:

• Python: نسخه 3.6 یا بالاتر
• Pip: مدیر بسته پایتون
• Flask: چارچوب وب پایتون
• Celery: سیستم صف پیام توزیع شده
• یک کارگزار پیام (Message Broker): مانند RabbitMQ یا Redis

برای نصب Flask و Celery، می‌توانید از Pip استفاده کنید:

pip install Flask Celery
 

همچنین، باید یک کارگزار پیام را نصب و پیکربندی کنید. در این مقاله، از RabbitMQ استفاده خواهیم کرد. برای نصب RabbitMQ، می‌توانید از دستورالعمل‌های موجود در وب‌سایت رسمی RabbitMQ استفاده کنید.

پیکربندی Celery

اولین قدم برای استفاده از Celery، پیکربندی آن است. برای این کار، یک فایل پیکربندی به نام `celeryconfig.py` ایجاد می‌کنیم و تنظیمات مربوط به Celery را در آن قرار می‌دهیم:

# celeryconfig.py

 broker_url = 'amqp://guest:guest@localhost:5672//'  # URL کارگزار پیام
 result_backend = 'redis://localhost:6379/0'  # URL ذخیره نتایج

 task_serializer = 'json'
 result_serializer = 'json'
 accept_content = ['json']
 timezone = 'Asia/Tehran'  # منطقه زمانی
 enable_utc = True
 

در این فایل، پارامتر `broker_url` را به URL کارگزار پیام خود (RabbitMQ) تنظیم می‌کنیم. همچنین، پارامتر `result_backend` را به URL محل ذخیره نتایج وظایف (Redis) تنظیم می‌کنیم. در اینجا ما از Redis برای ذخیره نتایج استفاده کرده‌ایم، اما می‌توانید از پایگاه داده‌های دیگر نیز استفاده کنید.

توضیح پارامترها:

• `broker_url`: آدرس URL کارگزار پیام (RabbitMQ در این مثال). این آدرس شامل نام کاربری، رمز عبور، هاست و پورت کارگزار پیام است.
• `result_backend`: آدرس URL محل ذخیره نتایج وظایف. در این مثال، از Redis برای ذخیره نتایج استفاده شده است.
• `task_serializer`: فرمت سریال‌سازی وظایف. `json` یک انتخاب رایج است.
• `result_serializer`: فرمت سریال‌سازی نتایج وظایف. `json` یک انتخاب رایج است.
• `accept_content`: لیستی از فرمت‌های محتوایی که Celery می‌تواند بپذیرد.
• `timezone`: منطقه زمانی برنامه.
• `enable_utc`: تعیین می‌کند که آیا Celery از زمان UTC استفاده کند یا خیر.

ادغام Celery با Flask

اکنون که Celery را پیکربندی کردیم، باید آن را با برنامه Flask خود ادغام کنیم. برای این کار، یک فایل به نام `app.py` ایجاد می‌کنیم و کد زیر را در آن قرار می‌دهیم:

# app.py

 from flask import Flask
 from celery import Celery

 def make_celery(app):
  celery = Celery(
   app.import_name,
   backend=app.config['CELERY_RESULT_BACKEND'],
   broker=app.config['CELERY_BROKER_URL']
  )
  celery.conf.update(app.config)

  class ContextTask(celery.Task):
   def __call__(self, *args, **kwargs):
    with app.app_context():
     return self.run(*args, **kwargs)

  celery.Task = ContextTask
  return celery

 app = Flask(__name__)
 app.config.update(
  CELERY_BROKER_URL='amqp://guest:guest@localhost:5672//',
  CELERY_RESULT_BACKEND='redis://localhost:6379/0'
 )
 celery = make_celery(app)

 @celery.task()
 def add_together(a, b):
  return a + b

 @app.route('/add//')
 def add(param1, param2):
  result = add_together.delay(param1, param2)
  return f"Task submitted. Result will be available later. Task ID: {result.id}"

 if __name__ == '__main__':
  app.run(debug=True)
 

در این فایل، ابتدا Flask و Celery را وارد می‌کنیم. سپس، یک تابع به نام `make_celery` ایجاد می‌کنیم که یک شی Celery را با تنظیمات Flask پیکربندی می‌کند. این تابع اطمینان حاصل می‌کند که Celery می‌تواند به زمینه برنامه Flask دسترسی داشته باشد.

در ادامه، یک شی Flask را ایجاد می‌کنیم و تنظیمات Celery را در پیکربندی Flask قرار می‌دهیم. سپس، تابع `make_celery` را فراخوانی می‌کنیم تا Celery را با Flask ادغام کنیم.

یک وظیفه Celery به نام `add_together` تعریف می‌کنیم که دو عدد را با هم جمع می‌کند. از دکوراتور `@celery.task()` برای تبدیل این تابع به یک وظیفه Celery استفاده می‌کنیم.

در نهایت، یک مسیر (route) به نام `/add//` تعریف می‌کنیم که دو عدد را از URL دریافت می‌کند و وظیفه `add_together` را به‌صورت غیرهمزمان اجرا می‌کند. تابع `delay` برای ارسال وظیفه به صف پیام Celery استفاده می‌شود. این تابع یک شی `AsyncResult` را برمی‌گرداند که می‌توانیم از آن برای بررسی وضعیت وظیفه و دریافت نتیجه استفاده کنیم.

توضیحات کد:

• `make_celery(app)`: این تابع یک شی Celery را ایجاد و پیکربندی می‌کند تا با برنامه Flask شما کار کند.
• `app.config.update(…)`: این خطوط پیکربندی Celery را در تنظیمات برنامه Flask ذخیره می‌کنند.
• `@celery.task()`: این دکوراتور یک تابع پایتون را به یک وظیفه Celery تبدیل می‌کند.
• `add_together.delay(param1, param2)`: این خط وظیفه `add_together` را به صف پیام Celery ارسال می‌کند تا در پس‌زمینه اجرا شود.
• `result.id`: این ویژگی شناسه یکتای وظیفه را برمی‌گرداند.

اجرای برنامه

برای اجرای برنامه، ابتدا باید کارگزار پیام (RabbitMQ) را اجرا کنید. سپس، یک ترمینال جدید باز کنید و دستور زیر را برای اجرای کارگر Celery وارد کنید:

celery -A app.celery worker --loglevel=info
 

در این دستور، `app.celery` به شی Celery تعریف شده در فایل `app.py` اشاره دارد. `–loglevel=info` سطح گزارش‌گیری را روی اطلاعات تنظیم می‌کند.

اکنون، یک ترمینال دیگر باز کنید و برنامه Flask را اجرا کنید:

python app.py
 

پس از اجرای برنامه، می‌توانید به آدرس `/add/2/3` در مرورگر خود مراجعه کنید. این آدرس وظیفه `add_together` را با پارامترهای 2 و 3 اجرا می‌کند و یک پیام مبنی بر ارسال وظیفه نمایش می‌دهد. کارگر Celery وظیفه را در پس‌زمینه پردازش می‌کند و نتیجه را در Redis ذخیره می‌کند.

دریافت نتایج وظایف

برای دریافت نتایج وظایف Celery، می‌توانیم از شی `AsyncResult` استفاده کنیم که توسط تابع `delay` برگردانده می‌شود. برای مثال، می‌توانیم کد زیر را به مسیر `/add` در فایل `app.py` اضافه کنیم:

from flask import Flask, jsonify
from celery import Celery
import time

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

app = Flask(__name__)
app.config.update(
    CELERY_BROKER_URL='amqp://guest:guest@localhost:5672//',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)
celery = make_celery(app)

@celery.task()
def add_together(a, b):
    time.sleep(5)  # شبیه‌سازی یک وظیفه زمان‌بر
    return a + b

@app.route('/add//')
def add(param1, param2):
    result = add_together.delay(param1, param2)
    return jsonify({'task_id': result.id})

@app.route('/result/')
def task_result(task_id):
    result = celery.AsyncResult(task_id)
    if result.ready():
        return jsonify({'result': result.get()})
    else:
        return jsonify({'status': 'pending'})

if __name__ == '__main__':
    app.run(debug=True)

در این کد، یک مسیر جدید به نام `/result/` تعریف کرده‌ایم که شناسه وظیفه را از URL دریافت می‌کند و وضعیت و نتیجه وظیفه را برمی‌گرداند. از تابع `ready()` برای بررسی اینکه آیا وظیفه به پایان رسیده است یا خیر استفاده می‌کنیم. اگر وظیفه به پایان رسیده باشد، از تابع `get()` برای دریافت نتیجه استفاده می‌کنیم. در غیر این صورت، یک پیام مبنی بر در حال انتظار بودن وظیفه برمی‌گردانیم.

برای استفاده از این مسیر، ابتدا به آدرس `/add/2/3` مراجعه کنید تا وظیفه `add_together` اجرا شود. سپس، شناسه وظیفه را از پاسخ JSON دریافت کنید و به آدرس `/result/` مراجعه کنید. اگر وظیفه به پایان رسیده باشد، نتیجه آن را مشاهده خواهید کرد. در غیر این صورت، یک پیام مبنی بر در حال انتظار بودن وظیفه مشاهده خواهید کرد.

مدیریت خطاها

Celery دارای مکانیسم‌های داخلی برای مدیریت خطاها است. اگر یک وظیفه با خطا مواجه شود، Celery می‌تواند آن را به‌طور خودکار دوباره تلاش کند. تعداد دفعات تلاش مجدد و تاخیر بین تلاش‌ها را می‌توان در تنظیمات Celery پیکربندی کرد.

برای مثال، می‌توانیم دکوراتور `@celery.task()` را با پارامترهای `bind` و `retry_backoff` پیکربندی کنیم تا در صورت بروز خطا، وظیفه به‌طور خودکار دوباره تلاش شود:

@celery.task(bind=True, retry_backoff=True, retry_kwargs={'max_retries': 3})
 def add_together(self, a, b):
  try:
   return a + b
  except Exception as e:
   self.retry(exc=e)

در این کد، پارامتر `bind=True` به این معنی است که شی وظیفه به عنوان اولین آرگومان به تابع ارسال می‌شود. پارامتر `retry_backoff=True` به این معنی است که تاخیر بین تلاش‌ها به طور تصاعدی افزایش می‌یابد. پارامتر `retry_kwargs={‘max_retries’: 3}` حداکثر تعداد دفعات تلاش مجدد را روی 3 تنظیم می‌کند.

در داخل تابع، یک بلوک `try…except` برای مدیریت خطاها اضافه کرده‌ایم. اگر خطایی رخ دهد، تابع `self.retry(exc=e)` را فراخوانی می‌کنیم تا وظیفه دوباره تلاش شود. پارامتر `exc=e` خطا را به Celery ارسال می‌کند تا بتواند آن را ثبت کند.

وظایف زمان‌بندی شده

Celery همچنین از وظایف زمان‌بندی شده پشتیبانی می‌کند. وظایف زمان‌بندی شده وظایفی هستند که در یک زمان مشخص یا با یک فاصله زمانی معین اجرا می‌شوند.

برای زمان‌بندی وظایف، می‌توانیم از Celery Beat استفاده کنیم. Celery Beat یک زمان‌بند است که وظایف را به Celery ارسال می‌کند. برای پیکربندی Celery Beat، باید یک فایل پیکربندی به نام `celerybeatconfig.py` ایجاد کنیم:

# celerybeatconfig.py

 from celery.schedules import crontab

 beat_schedule = {
  'add-every-minute': {
   'task': 'app.add_together',
   'schedule': crontab(minute='*/1'),
   'args': (16, 16)
  },
 }

در این فایل، یک دیکشنری به نام `beat_schedule` تعریف کرده‌ایم که شامل لیستی از وظایف زمان‌بندی شده است. هر وظیفه دارای یک نام، یک وظیفه، یک زمان‌بندی و آرگومان‌ها است.

در این مثال، یک وظیفه به نام `add-every-minute` تعریف کرده‌ایم که وظیفه `app.add_together` را هر دقیقه اجرا می‌کند. زمان‌بندی وظیفه با استفاده از یک عبارت crontab تعریف شده است. آرگومان‌های وظیفه (16 و 16) به وظیفه ارسال می‌شوند.

برای اجرای Celery Beat، دستور زیر را وارد کنید:

celery -A app.celery beat --loglevel=info

پس از اجرای Celery Beat، وظیفه `add_together` هر دقیقه اجرا می‌شود.

نتیجه‌گیری

در این مقاله، نحوه پیاده‌سازی وظایف پس‌زمینه در برنامه‌های Flask با استفاده از Celery را بررسی کردیم. Celery یک سیستم صف پیام توزیع شده قدرتمند است که به توسعه‌دهندگان اجازه می‌دهد وظایف را به‌صورت غیرهمزمان اجرا کنند. با استفاده از Celery، می‌توانیم پاسخگویی، مقیاس‌پذیری، انعطاف‌پذیری و قابلیت اطمینان برنامه‌های Flask خود را بهبود بخشیم.

ما در این مقاله به مباحث زیر پرداختیم:

• چرا از وظایف پس‌زمینه استفاده کنیم؟
• پیش‌نیازها
• پیکربندی Celery
• ادغام Celery با Flask
• اجرای برنامه
• دریافت نتایج وظایف
• مدیریت خطاها
• وظایف زمان‌بندی شده

امیدواریم این مقاله برای شما مفید بوده باشد. اگر سوالی دارید، می‌توانید در بخش نظرات مطرح کنید.

منابع

• Flask Documentation
• Celery Documentation
• RabbitMQ Website
• Redis Website

“`