Belajar Flask - Background Tasks

#Pendahuluan

Sebagian pekerjaan tidak boleh memperlambat response HTTP: mengirim email setelah registrasi, generate report Excel besar, memproses upload video, atau sinkronisasi dengan API pihak ketiga. Kalau dikerjakan di dalam view, pengguna menunggu berdetik-detik — atau request timeout. Episode 16 mengajarkan cara memindahkan pekerjaan ini keluar dari request: background tasks. Kita akan membandingkan tiga tingkat solusi — threading sederhana, RQ dengan Redis Queue, dan Celery untuk kebutuhan besar — lalu membangun praktik kirim email dan generate report.

#Mengapa Tidak Langsung di View?

Bayangkan endpoint POST /reports yang memproses 10.000 baris data. Kalau semua diproses di dalam view:

Anti-pattern - kerja berat di view

@app.post("/reports")
def create_report() -> tuple[dict, int]:
    rows = db.session.scalars(db.select(Order).limit(10000)).all()
    pdf = build_pdf(rows)          # makan waktu puluhan detik
    email_report(pdf)              # jaringan lambat lagi
    return api_ok({"status": "done"})

Pengguna menunggu; jika operasi gagal di tengah, response error dan proses tidak bisa diulang. Background task mengubahnya: view mengantrekan pekerjaan dan langsung merespons "diterima" (202 Accepted), lalu worker memproses di belakang.

#Tingkat 1: Threading untuk Tugas Ringan

Untuk tugas singkat (mengirim satu email, notifikasi) di aplikasi kecil, threading cukup:

Background task dengan threading

import threading
 
 
def send_welcome_email(user_email: str) -> None:
    # operasi jaringan yang lambat
    ...
 
 
@app.post("/register")
def register() -> tuple[dict, int]:
    # ... buat user ...
    threading.Thread(target=send_welcome_email, args=(email,)).start()
    return api_ok({"status": "ok"}), 201

Kekurangan pendekatan ini: tidak ada retry otomatis, tidak ada persistensi (task hilang saat server restart), dan thread tak terkelola bisa menumpuk. Cocok hanya untuk tugas kecil non-kritis di aplikasi kecil.

#Tingkat 2: RQ — Redis Queue

Untuk task yang butuh keandalan, gunakan RQ (Redis Queue): task masuk ke queue di Redis, worker memprosesnya, dan task bisa diulang saat gagal. Install dengan pip install "rq[redis]" dan pastikan Redis berjalan (redis-server).

Definisikan task dalam modul tersendiri (wajib: task harus bisa di-import worker):

tasks/email.py - task RQ

import redis
from rq import Queue
from app.extensions import db
from app.models.user import User
 
redis_conn = redis.from_url("redis://localhost:6379/0")
mail_queue = Queue("emails", connection=redis_conn)
 
 
def send_welcome_email(user_id: int) -> str:
    user = db.session.get(User, user_id)
    # kirim email...
    return f"Email terkirim ke {user.email}"

Dari view, antrekan task dan kembalikan status 202 Accepted:

View - enqueue task

from tasks.email import mail_queue, send_welcome_email
 
 
@app.post("/register")
def register() -> tuple[dict, int]:
    # ... buat user, user.id = 42 ...
    job = mail_queue.enqueue(send_welcome_email, user.id)
    return api_ok({"job_id": job.id}), 202

enqueue() menempatkan task ke queue; view langsung merespons. Worker dijalankan terpisah:

rq worker emails

Worker RQ memproses task berurutan, mendukung retry, dan task yang antre tetap aman di Redis meski server Flask restart.

#Tingkat 3: Celery — Untuk Skala Besar

Saat kebutuhan tumbuh — task terjadwal (cron), pipeline bertingkat, prioritas, multi-queue, monitoring — beralihlah ke Celery, task queue Python paling matang:

tasks/celery_app.py

from celery import Celery
 
celery_app = Celery("app", broker="redis://localhost:6379/0")
 
 
@celery_app.task(bind=True, max_retries=3)
def send_weekly_report(self, user_id: int) -> str:
    try:
        user = User.query.get(user_id)
        report = generate_report(user.id)
        send_email(user.email, report)
        return "ok"
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

Worker Celery dijalankan dengan celery -A tasks.celery_app worker. Keunggulan Celery: task scheduling (celery beat), retry dengan backoff, monitoring (Flower), dan batasan konkurensi per worker. Trade-off-nya: kompleksitas setup lebih tinggi — pilih saat RQ sudah terasa sempit.

#Praktik: Generate Report Asinkron

Kita gabungkan semuanya: endpoint menerima permintaan report, task dibuat, status dilacak:

tasks/reports.py - report asinkron

import io
from rq import Queue, Retry
import redis
from app.extensions import db
from app.models.order import Order
 
redis_conn = redis.from_url("redis://localhost:6379/0")
report_queue = Queue("reports", connection=redis_conn)
 
 
def build_and_email_report(user_id: int) -> str:
    rows = db.session.scalars(
        db.select(Order).where(Order.user_id == user_id)
    ).all()
    content = render_report(rows)          # kerja berat di sini
    write_report_to_disk(content)          # lalu simpan / kirim
    return f"Report user {user_id}: {len(rows)} baris"

View - enqueue dengan retry

job = report_queue.enqueue(
    build_and_email_report,
    user.id,
    retry=Retry(max=3, interval=[60, 300, 600]),
    job_timeout=600,
)

Retry(max=3, interval=[...]) mengulang task yang gagal dengan jeda bertingkat; job_timeout=600 memberi batas eksekusi maksimal. Client dapat memantau status via job.status (queued → started → finished/failed).

#Pola dan Pertimbangan

Beberapa pola yang perlu dipahami:

• Simpan state task: untuk UI, simpan job_id dan status di database agar halaman bisa menampilkan progres.
• Idempotency: task bisa dijalankan ulang (retry) — pastikan mengirim email/report dua kali tidak bermasalah, atau cek status sebelumnya.
• Monitoring: Redis menyimpan info job; untuk visibilitas lebih, pasang dashboard (RQ Dashboard / Flower untuk Celery).

#Common Pitfalls Background Tasks

• Task mengakses context Flask: worker tidak punya request context. Buat application context manual: with app.app_context(): di dalam task, atau buka session DB sendiri.
• target=f(...) tanpa args: fungsi langsung dieksekusi — background tidak pernah terjadi.
• Worker tidak jalan: task tetap "queued" selamanya. Selalu verifikasi rq worker aktif.
• Threading untuk task besar: tanpa queue, retry, dan persistensi — cepat habis saat beban naik.

#Penutup

Pada episode 16 ini, kalian telah memindahkan pekerjaan berat keluar dari request.

Inti yang harus dibawa pulang:

• Jangan kerjakan operasi lambat di dalam view — antrekan dan balas 202 Accepted.
• threading cukup untuk tugas ringan; RQ + Redis untuk keandalan; Celery untuk skala besar.
• Retry + job_timeout untuk keandalan; simpan status task agar UI bisa memantau.

Di episode 17 selanjutnya, kita mengeraskan aplikasi dari sisi keamanan: security best practices — XSS dan autoescape, CSRF, SQL injection di ORM, clickjacking, secure headers dengan Talisman, dan hardening menyeluruh. Sampai jumpa di episode 17!