Background Tasks dengan Celery + Redis di Flask

Di artikel ini, kita akan membahas cara mengimplementasikan Background Tasks di Flask menggunakan Celery dengan Redis sebagai message broker.

1. Mengapa Perlu Background Tasks?

Beberapa alasan utama:

Menghindari request timeout
Meningkatkan responsivitas aplikasi
Memproses tugas yang berat tanpa membebani web server
Menjalankan tugas secara asinkron dan terjadwal
Meningkatkan skalabilitas

Contoh tugas yang cocok dijalankan di background:

Mengirim email
Generate PDF / laporan
Resize dan optimasi gambar
Sinkronisasi data dengan layanan pihak ketiga
Proses machine learning / analisis data

2. Apa itu Celery?

Celery adalah distributed task queue yang sangat populer di ekosistem Python. Celery memungkinkan Anda menjalankan tugas secara asinkron di luar request cycle.

Komponen Utama Celery:

Komponen	Fungsi
Task	Fungsi Python yang akan dijalankan di background
Broker	Message queue (Redis / RabbitMQ)
Worker	Proses yang menjalankan task
Result Backend	Tempat menyimpan hasil task (opsional)
Beat	Scheduler untuk periodic task

3. Arsitektur Celery + Redis

1
Flask App
2
   ↓
3
   Celery Task (dipanggil secara async)
4
   ↓
5
Redis (Message Broker)
6
   ↓
7
Celery Worker (menjalankan task)
8
   ↓
9
Redis (Result Backend - opsional)

4. Instalasi

1
pip install celery redis

Jika menggunakan Flask, biasanya juga diperlukan:

1
pip install Flask-SQLAlchemy Flask-Mail

5. Konfigurasi Celery di Flask

Struktur Proyek yang Direkomendasikan

1
myapp/
2
├── app/
3
│   ├── __init__.py
4
│   ├── tasks.py
5
│   └── celery.py
6
├── celery_worker.py
7
└── run.py

`app/celery.py`

1
from celery import Celery
2

3
def make_celery(app):
4
    celery = Celery(
5
        app.import_name,
6
        backend=app.config['CELERY_RESULT_BACKEND'],
7
        broker=app.config['CELERY_BROKER_URL']
8
    )
9
    celery.conf.update(app.config)
10

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

16
    celery.Task = ContextTask
17
    return celery

`app/init.py`

1
from flask import Flask
2
from .celery import make_celery
3

4
app = Flask(__name__)
5
app.config.update(
6
    CELERY_BROKER_URL='redis://localhost:6379/0',
7
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
8
)
9

10
celery = make_celery(app)

6. Membuat Task

`app/tasks.py`

1
from . import celery
2
from time import sleep
3

4
@celery.task(bind=True, max_retries=3)
5
def send_welcome_email(self, user_id):
6
    try:
7
        # Logika mengirim email
8
        print(f"Sending welcome email to user {user_id}")
9
        sleep(5)  # Simulasi proses berat
10
        return f"Email sent to user {user_id}"
11
    except Exception as exc:
12
        raise self.retry(exc=exc, countdown=60)

7. Memanggil Task dari Flask

1
from flask import Flask, jsonify
2
from .tasks import send_welcome_email
3

4
@app.route('/register', methods=['POST'])
5
def register():
6
    # ... proses registrasi user ...
7

8
    # Panggil task secara asynchronous
9
    send_welcome_email.delay(user.id)
10

11
    return jsonify({"message": "User registered successfully"}), 201

Perbedaan pemanggilan:

.delay() → Asynchronous (paling umum)
.apply_async() → Bisa menambahkan countdown, eta, dll
() → Synchronous (jarang digunakan)

8. Menjalankan Celery Worker

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

Atau buat file celery_worker.py:

1
from app import celery
2

3
if __name__ == '__main__':
4
    celery.start()

Jalankan dengan:

1
python celery_worker.py

9. Monitoring dengan Flower

Flower adalah tool monitoring untuk Celery yang sangat berguna.

1
pip install flower

Jalankan:

1
celery -A app.celery flower

Akses di: http://localhost:5555

Flower menampilkan:

Task yang sedang berjalan
Task yang gagal
Statistik worker
Grafik performa

10. Periodic Tasks (Celery Beat)

Untuk menjalankan task secara terjadwal:

Konfigurasi Beat

1
app.config['CELERY_BEAT_SCHEDULE'] = {
2
    'send-daily-report': {
3
        'task': 'app.tasks.send_daily_report',
4
        'schedule': 86400.0,  # Setiap 24 jam
5
    },
6
}

Jalankan beat:

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

11. Error Handling & Retries

Celery menyediakan mekanisme retry yang powerful:

1
@celery.task(bind=True, max_retries=5, default_retry_delay=60)
2
def process_data(self, data_id):
3
    try:
4
        # proses data
5
    except Exception as exc:
6
        raise self.retry(exc=exc)

Anda juga bisa menggunakan autoretry_for:

1
from celery import Celery
2
from requests.exceptions import RequestException
3

4
@celery.task(autoretry_for=(RequestException,), retry_kwargs={'max_retries': 3})
5
def call_external_api():
6
    ...

12. Best Practices

Praktik	Rekomendasi
Gunakan `bind=True` jika butuh retry	Sangat disarankan
Jangan lakukan query database di task tanpa app context	Gunakan `with app.app_context()`
Gunakan `max_retries` dan `countdown`	Hindari spam retry
Monitor dengan Flower	Wajib di production
Gunakan Redis sebagai broker	Ringan dan cepat
Pisahkan task ke file terpisah	Lebih rapi
Gunakan `task_always_eager = True` di testing	Memudahkan testing

Kesimpulan

Celery + Redis adalah kombinasi yang sangat powerful untuk menangani background tasks di aplikasi Flask. Dengan menggunakan Celery, Anda bisa:

Membuat aplikasi lebih responsif
Menjalankan proses berat secara asinkron
Menjadwalkan tugas secara otomatis
Meningkatkan skalabilitas aplikasi

Meskipun ada sedikit kompleksitas dalam setup awal, manfaat jangka panjangnya sangat besar, terutama untuk aplikasi yang memiliki banyak proses berat.

Mulailah dengan tugas sederhana (seperti mengirim email), lalu kembangkan sesuai kebutuhan proyek Anda.

Thanks for reading!