Flask এবং Celery Integration (Task Queue)

Flask অ্যাপ্লিকেশনে Celery ব্যবহার করা হয় Asynchronous Task Queue হিসেবে, যেখানে আপনি দীর্ঘ-running বা ব্যাকগ্রাউন্ডে চলা কাজগুলি পরিচালনা করতে পারেন। Celery ব্যবহার করে আপনি Flask অ্যাপে বিভিন্ন ব্যাকগ্রাউন্ড টাস্ক যেমন ইমেইল পাঠানো, ডাটা প্রসেসিং, ফাইল প্রসেসিং, অথবা অন্য যেকোনো দীর্ঘ-running কাজ সমাধান করতে পারেন, যা ব্যবহারকারীর জন্য অপেক্ষার সময় কমিয়ে দেয়।

এই টিউটোরিয়ালে Flask এবং Celery এর ইন্টিগ্রেশন সম্পর্কে জানব।

১. Celery ইনস্টলেশন

প্রথমে, Flask অ্যাপ্লিকেশন এবং Celery এর জন্য প্রয়োজনীয় প্যাকেজগুলো ইনস্টল করতে হবে।

pip install Flask Celery redis

এখানে:

• Celery: এটি ব্যাকগ্রাউন্ড টাস্ক পরিচালনা করার জন্য ব্যবহৃত হয়।
• redis: Redis হল একটি ইন-মেমরি ডেটাবেস যা Celery কাজের কিউ (Queue) হিসেবে ব্যবহৃত হয়। Redis একটি জনপ্রিয় ব্রোকার হিসেবে ব্যবহৃত হয় Celery এর সাথে।

২. Flask এবং Celery কনফিগারেশন

Flask অ্যাপ্লিকেশন এবং Celery ইন্টিগ্রেট করার জন্য আপনাকে Flask কনফিগারেশনে কিছু সেটিংস যোগ করতে হবে। Celery কাজ করার জন্য একটি Broker এবং Backend প্রয়োজন।

উদাহরণ Flask অ্যাপ কনফিগারেশন:

from flask import Flask
from celery import Celery

app = Flask(__name__)

# Celery কনফিগারেশন
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'  # Redis ব্রোকার
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'  # Redis Backend

# Celery অ্যাপ ইনিশিয়ালাইজ করা
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@app.route('/')
def index():
    return 'Flask এবং Celery ইন্টিগ্রেশন সফল!'

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

এখানে:

• CELERY_BROKER_URL: Redis সার্ভারের URL যেখানে Celery কাজের কিউ রাখা হবে।
• CELERY_RESULT_BACKEND: এটি Celery এর কাজের রেজাল্ট বা আউটপুট যেখানে সংরক্ষিত হবে।

৩. Celery Task তৈরি করা

Flask অ্যাপ্লিকেশনে ব্যাকগ্রাউন্ড টাস্ক তৈরি করতে আপনাকে Celery টাস্ক ডেকোরেটর ব্যবহার করতে হবে। এখানে একটি উদাহরণ দেওয়া হলো যেখানে একটি দীর্ঘ-running কাজ (যেমন, স্লিপ করা) করা হচ্ছে।

উদাহরণ: Celery Task

from celery import Celery
import time

app = Flask(__name__)

# Celery কনফিগারেশন
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

# একটি সহজ Celery Task
@celery.task
def long_task():
    time.sleep(5)  # 5 সেকেন্ডের জন্য স্লিপ
    return 'Task Completed!'

@app.route('/start_task')
def start_task():
    # Celery Task চালানো
    long_task.apply_async()
    return 'Task Started!'

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

এখানে:

• @celery.task: এটি একটি Celery task ডেকোরেটর যা ফাংশনকে একটি ব্যাকগ্রাউন্ড টাস্কে রূপান্তরিত করে।
• long_task.apply_async(): এটি ব্যাকগ্রাউন্ডে long_task ফাংশন চালাতে ব্যবহৃত হয়।

ব্যাখ্যা:

• long_task() ফাংশনটি 5 সেকেন্ডের জন্য স্লিপ করবে এবং তার পরে "Task Completed!" রিটার্ন করবে।
• apply_async(): এটি সিঙ্ক্রোনাস না হয়ে অ্যাসিঙ্ক্রোনাসভাবে কাজটি চালাবে।

৪. Celery Worker চালানো

Celery টাস্ক রান করার জন্য আপনাকে Celery worker চালাতে হবে। এটি Flask অ্যাপের বাইরে একটি আলাদা প্রক্রিয়া হিসেবে চলবে।

আপনি কমান্ড লাইন থেকে নিচের কমান্ড দিয়ে Celery worker চালাতে পারেন:

celery -A app.celery worker

এখানে:

• -A app.celery: এটি Flask অ্যাপের মধ্যে Celery ইনস্ট্যান্সকে নির্দেশ দেয়।
• worker: এটি Celery worker চালানোর জন্য নির্দেশ।

Celery worker সফলভাবে চালু হলে, এটি ব্যাকগ্রাউন্ড টাস্কগুলি প্রসেস করবে এবং এগুলির আউটপুট Redis Backend এ সংরক্ষণ করবে।

৫. Celery এবং Flask-এর মধ্যে ফলাফল সংগ্রহ করা

যেহেতু Celery ব্যাকগ্রাউন্ডে কাজ করছে, আপনি সেই কাজের ফলাফল সংগ্রহ করতে পারেন। Celery-তে কাজের ফলাফল সংগ্রহের জন্য AsyncResult ব্যবহার করা হয়।

উদাহরণ: ফলাফল সংগ্রহ করা

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

app = Flask(__name__)

# Celery কনফিগারেশন
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

# Celery Task
@celery.task
def long_task():
    time.sleep(5)
    return 'Task Completed'

@app.route('/start_task')
def start_task():
    task = long_task.apply_async()  # Celery task শুরু করা
    return jsonify({"task_id": task.id})

@app.route('/get_task_result/<task_id>')
def get_task_result(task_id):
    task = long_task.AsyncResult(task_id)  # Task ফলাফল নেওয়া
    if task.state == 'SUCCESS':
        return jsonify({"result": task.result})  # কাজ সফল হলে ফলাফল দেখান
    elif task.state == 'PENDING':
        return jsonify({"status": "Task is still processing"})
    else:
        return jsonify({"status": "Task failed or is invalid"})

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

এখানে:

• apply_async(): এটি Celery টাস্ক শুরু করার জন্য ব্যবহৃত হয় এবং একটি task id প্রদান করে।
• AsyncResult: এটি টাস্কের ফলাফল বা স্টেট যাচাই করার জন্য ব্যবহৃত হয়।

ফলাফল সংগ্রহ করার URL:

• /start_task: নতুন টাস্ক শুরু করার জন্য URL।
• /get_task_result/<task_id>: টাস্কের ফলাফল সংগ্রহ করার জন্য URL।

৬. Celery এবং Flask Integration Benefits

• Asynchronous Task Processing: Celery ব্যবহার করে Flask অ্যাপে ব্যাকগ্রাউন্ড টাস্ক চলানো যায়, যা ইউজারের জন্য সঠিকভাবে রেসপন্স দেওয়ার সময়কে কমায়।
• Scalability: Celery একাধিক কাজের জন্য সঠিকভাবে স্কেল করা যায়।
• Retry Mechanism: Celery ত্রুটি হলে কাজ পুনরায় চেষ্টা করার জন্য কনফিগার করা যেতে পারে।

Flask-এ Celery ইন্টিগ্রেশন অ্যাপ্লিকেশনের পারফরম্যান্স উন্নত করতে সহায়ক, বিশেষ করে যখন আপনার অ্যাপ্লিকেশনকে দীর্ঘ-running বা ব্যাকগ্রাউন্ড টাস্ক সমাধান করতে হয়। Celery ব্যবহার করে আপনি অ্যাসিঙ্ক্রোনাস কাজ করতে পারেন, যা ব্যবহারকারীদের জন্য দ্রুত এবং কার্যকর রেসপন্স প্রদান করে। Flask এবং Celery এর মাধ্যমে অ্যাপ্লিকেশনকে আরো স্কেলেবল এবং দ্রুত কার্যকরী করে তুলতে পারেন।

Flask-এ Celery ব্যবহার করা হয় asynchronous task processing বা background task processing পরিচালনা করার জন্য। Celery একটি অত্যন্ত শক্তিশালী এবং জনপ্রিয় টুল, যা Flask অ্যাপ্লিকেশনে ব্যাকগ্রাউন্ড টাস্ক পরিচালনা করতে সাহায্য করে, যেমন ইমেইল পাঠানো, ফাইল প্রসেসিং, বা বড় ডেটা প্রসেসিং।

Celery সাধারণত Task Queue ম্যানেজমেন্টের জন্য ব্যবহৃত হয়, যেখানে টাস্কগুলো একটি Queue তে রাখা হয় এবং worker দ্বারা প্রসেস করা হয়। এটি Flask অ্যাপ্লিকেশনে ফিচারগুলোর কার্যকারিতা উন্নত করতে সাহায্য করে, যেমন দীর্ঘ সময় নেওয়া কাজগুলিকে প্রধান থ্রেডে ব্লক না করে ব্যাকগ্রাউন্ডে চালানো।

এই টিউটোরিয়ালে আমরা Flask অ্যাপে Celery ব্যবহার করে Task Queue Management কিভাবে করতে হয় তা দেখব।

১. Celery সেটআপ

Celery কাজ করার জন্য কিছু নির্দিষ্ট কনফিগারেশন এবং ব্যবস্থাপনা প্রয়োজন। সাধারণত এটি একটি message broker যেমন RabbitMQ বা Redis ব্যবহার করে।

ধাপ ১: Celery ইনস্টলেশন

প্রথমে Flask এবং Celery ইনস্টল করতে হবে। আপনি pip ব্যবহার করে Flask এবং Celery ইনস্টল করতে পারেন:

pip install flask
pip install celery
pip install redis  # Redis সার্ভার ব্যবহার করতে হলে

Redis সার্ভার ব্যবহারের জন্য, আপনি আপনার লোকাল মেশিনে Redis ইনস্টল করতে হবে। অথবা, Docker ব্যবহার করে Redis চালু করতে পারেন।

docker run -p 6379:6379 redis

ধাপ ২: Flask অ্যাপ্লিকেশনে Celery ইন্টিগ্রেশন

Flask অ্যাপ্লিকেশনে Celery কনফিগার করতে Celery এবং Redis এর কনফিগারেশন যোগ করা হয়।

from flask import Flask, jsonify
from celery import Celery

app = Flask(__name__)

# Celery কনফিগারেশন
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'  # Redis URL
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

# Celery ইনস্ট্যান্স তৈরি করা
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@app.route('/process_task')
def process_task():
    task = long_task.apply_async()  # ব্যাকগ্রাউন্ড টাস্ক চালানো
    return jsonify({"task_id": task.id}), 202

@celery.task
def long_task():
    import time
    time.sleep(10)  # 10 সেকেন্ড স্লিপ করে দীর্ঘ সময় নেওয়া কাজ
    return 'Task completed!'

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

এখানে:

• CELERY_BROKER_URL: এটি Redis সার্ভারের URL।
• CELERY_RESULT_BACKEND: এটি যেখানে টাস্কের ফলাফল সংরক্ষণ করবে, সাধারণত Redis বা অন্য ডেটাবেস ব্যবহার করা হয়।
• long_task(): এটি একটি Celery task যা ব্যাকগ্রাউন্ডে রান করবে।

২. Task Queue Management

Celery অ্যাপ্লিকেশনটি মূলত টাস্কগুলো কিউতে রেখে সেগুলোকে প্রসেস করতে সাহায্য করে। যখন একটি টাস্ক সাবমিট করা হয়, তখন Celery সেই টাস্ককে কিউতে রাখে এবং worker দ্বারা প্রসেস করা হয়। এটি অ্যাসিঙ্ক্রোনাসভাবে কাজ করে, অর্থাৎ Flask অ্যাপ্লিকেশন ব্যাকগ্রাউন্ডে টাস্ক প্রক্রিয়াকরণের জন্য থামবে না।

উদাহরণ:

আপনি /process_task URL এ গিয়ে long_task টাস্কটিকে asynchronous ভাবে শুরু করতে পারেন।

curl http://127.0.0.1:5000/process_task

এটি একটি HTTP রিকোয়েস্ট পাঠাবে এবং Celery টাস্কটি Redis কিউতে রেখে দিতে হবে।

৩. Celery Worker চালানো

Celery-তে টাস্ক প্রসেস করার জন্য একটি worker চালানো দরকার। এটি ফ্লাস্ক অ্যাপের বাইরে চলে এবং Redis কিউ থেকে টাস্কগুলি প্রসেস করে।

Celery worker চালাতে:

celery -A app.celery worker

এখানে:

• -A app.celery: এটি Flask অ্যাপের celery ইনস্ট্যান্সকে নির্দেশ দেয়।
• worker: Celery worker প্রসেস শুরু করে, যা কিউ থেকে টাস্ক নেবে এবং প্রসেস করবে।

এখন, যদি আপনি /process_task URL এ একটি রিকোয়েস্ট পাঠান, তাহলে Celery worker এই টাস্কটিকে Redis কিউ থেকে নিয়ে প্রসেস করবে।

৪. Celery Result Retrieval

Celery-তে আপনি একটি টাস্কের ফলাফল সংগ্রহ করতে পারেন। এটি AsyncResult ব্যবহার করে করা হয়, যার মাধ্যমে আপনি টাস্কের স্ট্যাটাস এবং ফলাফল জানতে পারবেন।

@app.route('/task_status/<task_id>')
def task_status(task_id):
    task = long_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        return jsonify({"status": "Task is pending"})
    elif task.state != 'FAILURE':
        return jsonify({"status": task.state, "result": task.result})
    else:
        return jsonify({"status": "Task failed"})

এখানে:

• AsyncResult: এটি টাস্কের বর্তমান স্ট্যাটাস এবং ফলাফল ফেচ করতে ব্যবহৃত হয়।

৫. Flask এবং Celery এর মাধ্যমে কনক্রিট টাস্কের উদাহরণ

ধরা যাক, আপনি একটি বড় ডেটা প্রক্রিয়া চালাতে চান যা দীর্ঘ সময় নেয়, যেমন ইমেইল পাঠানো বা রিপোর্ট জেনারেশন। আপনি সেই কাজটি Celery দিয়ে ব্যাকগ্রাউন্ডে চালিয়ে দিতে পারেন, যাতে Flask অ্যাপ্লিকেশন থামবে না এবং ব্যবহারকারীকে দ্রুত সাড়া দেওয়া যায়।

উদাহরণ:

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

app = Flask(__name__)

# Celery কনফিগারেশন
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/start_task')
def start_task():
    task = background_task.apply_async()  # টাস্ক শুরু
    return f'Task started with id {task.id}'

@celery.task
def background_task():
    time.sleep(30)  # 30 সেকেন্ডের জন্য স্লিপ করা
    return 'Task completed'

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

এখানে:

• background_task(): এই টাস্কটি Celery worker দ্বারা 30 সেকেন্ডের জন্য প্রসেস হবে।
• apply_async(): এটি টাস্কটিকে asynchronously 실행 করার জন্য ব্যবহৃত হয়।

Flask এবং Celery এর সংমিশ্রণ ওয়েব অ্যাপ্লিকেশনগুলোতে background task management বা asynchronous task processing পরিচালনা করার জন্য খুবই শক্তিশালী এবং কার্যকরী। Celery দিয়ে আপনি দীর্ঘ সময় নেওয়া কাজগুলোকে ব্যাকগ্রাউন্ডে রাখতে পারেন, যেমন ইমেইল পাঠানো, রিপোর্ট জেনারেশন, বা বড় ডেটা প্রসেসিং। Celery worker এবং Redis ব্যবহারের মাধ্যমে আপনার অ্যাপ্লিকেশন দ্রুত এবং স্কেলেবল হয়, যা ব্যবহারকারীদের জন্য উন্নত অভিজ্ঞতা নিশ্চিত করে।

Flask-এ Celery ইন্টিগ্রেশন ব্যবহার করে ব্যাকগ্রাউন্ড টাস্ক চালানো যেতে পারে, যা আপনার অ্যাপ্লিকেশনকে দীর্ঘ চলমান কার্যক্রম বা সময়সাপেক্ষ কাজগুলো ব্যাকগ্রাউন্ডে চালানোর সুবিধা দেয়। Celery একটি শক্তিশালী, অ্যাসিঙ্ক্রোনাস টাস্ক কিউ ব্যবস্থাপনা সিস্টেম, যা ফ্লাস্কের মতো ওয়েব ফ্রেমওয়ার্কের সাথে সহজে কাজ করে।

এই টিউটোরিয়ালে, আমরা Flask এবং Celery ইন্টিগ্রেট করার পদ্ধতি দেখব।

১. Celery ইনস্টলেশন

প্রথমে, Flask অ্যাপ্লিকেশনে Celery যোগ করার জন্য Celery এবং Redis (কিউ এবং ফলাফল ব্যাকস্টোর হিসাবে) ইনস্টল করতে হবে। Redis হচ্ছে সবচেয়ে জনপ্রিয় ব্রোকার Celery-এর জন্য।

pip install celery redis

• Celery: Task queue ম্যানেজমেন্টের জন্য।
• Redis: Celery এর জন্য ব্রোকার হিসেবে ব্যবহৃত হয়।

২. Flask অ্যাপে Celery কনফিগারেশন

এখন, Flask অ্যাপ্লিকেশনে Celery ইন্টিগ্রেট করার জন্য কনফিগারেশন সেট করতে হবে।

উদাহরণ: Flask এবং Celery কনফিগারেশন

from flask import Flask
from celery import Celery

app = Flask(__name__)

# Celery কনফিগারেশন
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'  # Redis ব্রোকার
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'  # Redis ফলাফল ব্যাকস্টোর

# Celery ইনস্ট্যান্স তৈরি
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@app.route('/')
def index():
    return 'Flask Celery Integration Example'

এখানে:

• CELERY_BROKER_URL: এটি Redis সার্ভারের URL, যা Celery টাস্ক ম্যানেজমেন্টের জন্য ব্যবহৃত হয়।
• CELERY_RESULT_BACKEND: এটি Redis সার্ভারে টাস্ক ফলাফল সংরক্ষণ করতে ব্যবহৃত হয়।

৩. Flask অ্যাপে Celery Task তৈরি করা

Flask অ্যাপের মধ্যে Celery Task তৈরি করতে আমরা একটি ফাংশন ব্যবহার করব যা ব্যাকগ্রাউন্ডে চলবে।

উদাহরণ: একটি সিম্পল Celery Task

# Celery টাস্ক তৈরি
@celery.task
def add_numbers(a, b):
    return a + b

এখানে:

• add_numbers: এটি একটি সিম্পল অ্যাসিঙ্ক্রোনাস ফাংশন যা দুইটি সংখ্যা যোগ করবে এবং ফলাফল ফেরত দেবে।

৪. Celery Task কল করা

Celery টাস্ক অ্যাসিঙ্ক্রোনাসভাবে কল করা হয়। Flask অ্যাপ্লিকেশন থেকে এই টাস্ক কল করা যাবে এবং টাস্কটি ব্যাকগ্রাউন্ডে রান করবে।

উদাহরণ: Celery Task কল করা

@app.route('/add')
def add():
    task = add_numbers.apply_async(args=[5, 10])  # অ্যাসিঙ্ক্রোনাস টাস্ক কল
    return f"Task started! Task ID: {task.id}"

এখানে:

• apply_async(): এটি Celery টাস্ককে অ্যাসিঙ্ক্রোনাসভাবে রান করানোর জন্য ব্যবহৃত হয়।
• task.id: এটি টাস্কের ID যা পরে ট্র্যাক করা যেতে পারে।

৫. Celery Task এর ফলাফল চেক করা

Celery Task-এর ফলাফল ট্র্যাক করা যায়। টাস্ক সম্পন্ন হওয়ার পর আপনি সেই ফলাফল পেতে পারেন।

উদাহরণ: টাস্কের ফলাফল সংগ্রহ করা

@app.route('/result/<task_id>')
def get_result(task_id):
    task = celery.AsyncResult(task_id)
    if task.state == 'SUCCESS':
        return f'Task result: {task.result}'
    else:
        return f'Task state: {task.state}'

এখানে:

• AsyncResult: এটি Celery টাস্কের স্টেট এবং ফলাফল ট্র্যাক করতে ব্যবহৃত হয়।
• task.state: এটি টাস্কের স্টেট দেখাবে (যেমন, SUCCESS, PENDING, FAILURE)।
• task.result: এটি টাস্কের ফলাফল প্রদান করবে যদি টাস্ক সফলভাবে সম্পন্ন হয়।

৬. Flask অ্যাপ্লিকেশন চালানো

Flask অ্যাপ্লিকেশন চালানোর জন্য সাধারণ Flask কমান্ড ব্যবহার করা হয়।

flask run

এখন, Flask অ্যাপ্লিকেশনের সাথে Celery একত্রে কাজ করবে। Celery টাস্কগুলোর প্রসেসিং হবে এবং Flask অ্যাপ্লিকেশন টাস্কের ফলাফল ট্র্যাক করবে।

৭. Celery Worker চালানো

Celery টাস্কগুলো সম্পন্ন করতে Celery Worker চালাতে হবে। নিচের কমান্ড দিয়ে Celery Worker চালানো যাবে:

celery -A app.celery worker

এখানে:

• -A app.celery: app.celery Celery ইনস্ট্যান্সটি নির্দেশ করে, যা Flask অ্যাপের মধ্যে সেটআপ করা হয়েছে।
• worker: এটি Celery Worker চালাতে ব্যবহৃত হয়, যা ব্যাকগ্রাউন্ড টাস্ক প্রসেস করবে।

৮. Celery Beat (Periodic Task Scheduling)

Flask-এ Celery-কে periodic tasks (নির্দিষ্ট সময় অন্তর টাস্ক চালানো) পরিচালনা করতে Celery Beat ব্যবহার করা হয়। এটি সময়ে সময়ে বা নির্দিষ্ট সময় অন্তর টাস্ক চালাতে ব্যবহৃত হয়।

উদাহরণ: Celery Beat কনফিগারেশন

from celery import Celery
from celery.schedules import crontab

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])

# Celery Beat কনফিগারেশন
celery.conf.beat_schedule = {
    'send-email-every-minute': {
        'task': 'app.send_email',
        'schedule': crontab(minute='*/1'),  # প্রতি ১ মিনিট পর পর ইমেইল পাঠানো
    }
}

# টাস্ক সেট করা
@celery.task
def send_email():
    # ইমেইল পাঠানোর কোড এখানে থাকবে
    pass

এখানে:

• crontab(minute='*/1'): এটি প্রতি ১ মিনিটে টাস্ক চালানোর জন্য কনফিগার করা হয়েছে।

Flask এবং Celery ইন্টিগ্রেশন ব্যাকগ্রাউন্ড টাস্ক এবং অ্যাসিঙ্ক্রোনাস প্রসেসিংয়ের জন্য অত্যন্ত কার্যকরী। Flask-এ Celery ব্যবহার করে আপনি টাইম-নিবদ্ধ টাস্ক, ব্যাকগ্রাউন্ডে ফাইল প্রসেসিং, ইমেইল পাঠানো, ডেটাবেস আপডেট এবং অন্যান্য সময়সাপেক্ষ কাজ পরিচালনা করতে পারেন। Celery Beat ব্যবহার করে আপনি নিয়মিত এবং পরিকল্পিত টাস্কগুলোও পরিচালনা করতে পারেন। Flask এবং Celery এর সমন্বয় আপনার অ্যাপ্লিকেশনে কার্যকারিতা এবং স্কেলেবিলিটি যোগ করতে সহায়তা করবে।

Flask অ্যাপ্লিকেশনগুলি সাধারণত ওয়েব রিকোয়েস্টের উপর ভিত্তি করে কাজ করে। কিন্তু কিছু কাজ রয়েছে যা সময় নেয়, যেমন বড় ডেটাবেস অপারেশন, ফাইল প্রসেসিং, অথবা তৃতীয় পক্ষের সেবা (API) থেকে তথ্য সংগ্রহ। এই ধরনের কাজগুলি সাধারণত সিঙ্ক্রোনাসভাবে (একই থ্রেডে) চলে, যার ফলে ব্যবহারকারীর রিকোয়েস্ট প্রতিস্থাপিত হতে পারে এবং অ্যাপ্লিকেশনের পারফরম্যান্স খারাপ হতে পারে।

এক্ষেত্রে Background Tasks এবং Delayed Execution একটি কার্যকরী সমাধান হতে পারে। Flask-এ এই ধরনের কাজ পরিচালনা করতে কয়েকটি পদ্ধতি রয়েছে, যেমন Celery বা Flask-Executor। এখানে আমরা Celery ব্যবহার করে ব্যাকগ্রাউন্ড টাস্ক এবং দেরি করা এক্সিকিউশন (Delayed Execution) দেখব।

১. Flask-এ Background Task: Celery ব্যবহার করা

Celery হলো একটি জনপ্রিয় পাইথন লাইব্রেরি, যা Flask অ্যাপ্লিকেশনের জন্য ব্যাকগ্রাউন্ড টাস্ক এবং প্যারালাল প্রসেসিং পরিচালনা করতে ব্যবহৃত হয়। এটি দীর্ঘ-running কাজগুলি ব্যাকগ্রাউন্ডে পরিচালনা করে, যাতে মূল অ্যাপ্লিকেশন থ্রেড ব্লক না হয়।

ধাপ ১: Celery ইনস্টলেশন

প্রথমে Celery ইনস্টল করতে হবে:

pip install celery

ধাপ ২: Flask অ্যাপে Celery সেটআপ

Flask অ্যাপ্লিকেশন এবং Celery ইন্টিগ্রেট করতে নিচের কোড ব্যবহার করুন:

from flask import Flask, jsonify
from celery import Celery

# Flask অ্যাপ সেটআপ
app = Flask(__name__)

# Celery কনফিগারেশন
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'  # Redis ব্যবহার করে ব্রোকার URL
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'  # Redis ব্যবহার করে রেজাল্ট ব্যাকএন্ড

# Celery ইনিশিয়ালাইজ করা
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

# ব্যাকগ্রাউন্ড টাস্ক
@celery.task
def long_task():
    import time
    time.sleep(10)  # ১০ সেকেন্ড স্লিপ (লং রানিং কাজের উদাহরণ)
    return 'Task Completed!'

@app.route('/start_task', methods=['GET'])
def start_task():
    # ব্যাকগ্রাউন্ড টাস্ক শুরু করা
    task = long_task.apply_async()  # টাস্ক অ্যাসিঙ্ক্রোনাসভাবে শুরু
    return jsonify({"task_id": task.id}), 202  # টাস্ক আইডি ফেরত

@app.route('/task_status/<task_id>', methods=['GET'])
def task_status(task_id):
    task = long_task.AsyncResult(task_id)  # টাস্কের স্ট্যাটাস চেক
    if task.state == 'PENDING':
        return jsonify({"status": "Task is still pending"}), 200
    elif task.state == 'SUCCESS':
        return jsonify({"status": "Task completed!", "result": task.result}), 200
    elif task.state == 'FAILURE':
        return jsonify({"status": "Task failed"}), 500
    else:
        return jsonify({"status": "Unknown"}), 500

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

এখানে:

• CELERY_BROKER_URL: Redis সার্ভার URL যেখানে Celery টাস্কগুলো ব্রোকার হিসাবে কাজ করবে।
• long_task: একটি ব্যাকগ্রাউন্ড টাস্ক, যা ১০ সেকেন্ড স্লিপ করে পরে "Task Completed!" বার্তা ফেরত দেয়।
• apply_async(): এটি Celery টাস্ককে অ্যাসিঙ্ক্রোনাসভাবে চালু করে।
• AsyncResult(): এটি টাস্কের স্ট্যাটাস এবং ফলাফল চেক করতে ব্যবহৃত হয়।

২. Celery Worker চালানো

Celery অ্যাপ চালানোর জন্য আপনাকে Celery Worker চালাতে হবে। এটি ব্যাকগ্রাউন্ড টাস্ক পরিচালনা করবে। নিচের কমান্ড ব্যবহার করে Celery Worker চালাতে হবে:

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

এখানে:

• your_flask_app_name: Flask অ্যাপের ফাইলের নাম।
• --loglevel=info: Celery Worker এর লগ লেভেল।

৩. Flask-এ Background Task Testing

Flask অ্যাপ চালানোর পর, http://127.0.0.1:5000/start_task URL এ গিয়ে একটি ব্যাকগ্রাউন্ড টাস্ক শুরু করতে পারেন। টাস্কের স্ট্যাটাস জানতে /task_status/<task_id> রাউটটি ব্যবহার করুন।

• /start_task: ব্যাকগ্রাউন্ড টাস্ক শুরু করবে এবং একটি টাস্ক আইডি ফেরত দেবে।
• /task_status/<task_id>: টাস্কের স্ট্যাটাস এবং ফলাফল চেক করবে।

৪. Flask Executor (Delayed Execution)

Flask-এ কিছু কাজ দেরিতে (Delayed Execution) সম্পন্ন করতে Flask-Executor ব্যবহার করা যেতে পারে। এটি একটি সাধারণ এবং দ্রুত পদ্ধতি যারা Celery ব্যবহার করতে চান না।

ধাপ ১: Flask-Executor ইনস্টল করা

pip install flask-executor

ধাপ ২: Flask-Executor সেটআপ

from flask import Flask
from flask_executor import Executor
import time

app = Flask(__name__)

# Executor সেটআপ
executor = Executor(app)

def long_task():
    time.sleep(10)  # লং রানিং কাজের উদাহরণ
    print("Task Completed!")

@app.route('/start_task', methods=['GET'])
def start_task():
    executor.submit(long_task)  # ব্যাকগ্রাউন্ডে টাস্ক শুরু করা
    return 'Task started in background!'

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

এখানে:

• executor.submit(long_task): এটি ব্যাকগ্রাউন্ডে long_task() ফাংশন চালায়, যাতে ব্যবহারকারী রিকোয়েস্ট সেরে ফেলতে পারে এবং টাস্কটি আলাদা থ্রেডে চলে।

৫. Flask-এ Background Task এবং Delayed Execution এর সুবিধা

• ইউজার এক্সপেরিয়েন্স উন্নত: ব্যাকগ্রাউন্ড টাস্ক ব্যবহার করে ইউজারের ওয়েব রিকোয়েস্ট ব্লকিং রোধ করা যায়।
• রিসোর্স ব্যবস্থাপনা: বড় ও সময়সাপেক্ষ কাজগুলো ব্যাকগ্রাউন্ডে চলে, মূল অ্যাপ্লিকেশন দ্রুত কার্যক্রম সম্পন্ন করতে পারে।
• অ্যাপ্লিকেশন পারফরম্যান্স: Flask অ্যাপ্লিকেশনটি অধিক কার্যকরী হতে পারে যখন স্লো বা সময়সাপেক্ষ কাজগুলি আলাদা থ্রেডে চলে।

Background Tasks এবং Delayed Execution Flask অ্যাপ্লিকেশনে দীর্ঘ-running কাজ এবং প্রক্রিয়াগুলির কার্যকরী ব্যবস্থাপনা করতে সাহায্য করে। Flask-এ Celery ব্যবহার করে ব্যাকগ্রাউন্ড টাস্ক পরিচালনা করা যায়, অথবা সহজ পদ্ধতি হিসেবে Flask-Executor ব্যবহার করা যেতে পারে। এই ফিচারগুলো অ্যাপ্লিকেশন পারফরম্যান্স এবং ইউজার এক্সপেরিয়েন্স উন্নত করতে সহায়তা করে।