Caching and Background Tasks (Optional)

Introduction

Some work is too slow for a single HTTP request—generating reports, sending bulk email, or hitting slow external APIs. Caching stores expensive results for reuse; background tasks move long jobs off the request thread. This chapter introduces Flask-Caching with Redis and Celery concepts—optional until your app needs them.

Prerequisites

When to Cache

ScenarioCache?
Homepage queried 1000×/min with same HTMLYes
User-specific dashboardShort TTL or no cache
POST that creates ordersNo (invalidate instead)

Redis concepts overlap with Spring Boot Redis integration—same server, different client library.

Flask-Caching Setup

bash
pip install Flask-Caching redis

app/config.py:

python
class DevelopmentConfig(Config):
    CACHE_TYPE = "SimpleCache"
    CACHE_DEFAULT_TIMEOUT = 300
 
class ProductionConfig(Config):
    CACHE_TYPE = "RedisCache"
    CACHE_REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")

app/extensions.py:

python
from flask_caching import Cache
 
cache = Cache()

app/__init__.py:

python
from app.extensions import cache
 
def create_app(...):
    ...
    cache.init_app(app)
    ...

Cache a View

python
from app.extensions import cache
 
@main_bp.route("/stats")
@cache.cached(timeout=60, query_string=True)
def stats():
    return {"visits": compute_expensive_stats()}

query_string=True includes query params in cache key.

Manual key:

python
@main_bp.route("/user/<int:user_id>/summary")
def user_summary(user_id):
    cache_key = f"user_summary:{user_id}"
    data = cache.get(cache_key)
    if data is None:
        data = build_summary(user_id)
        cache.set(cache_key, data, timeout=120)
    return data

Invalidate Cache

After data changes:

python
cache.delete(f"user_summary:{user_id}")
cache.delete_memoized(stats)

Pattern: cache-aside—app reads cache, on miss loads DB, writes cache.

Do Not Block Requests on Heavy Work

Bad:

python
@main_bp.route("/report")
def report():
    send_500_emails()
    generate_pdf()
    return "Done"

User waits minutes; worker may timeout.

Good: enqueue task, return 202 Accepted:

python
@main_bp.route("/report", methods=["POST"])
def request_report():
    generate_report_task.delay(current_user.id)
    return jsonify({"message": "Report queued"}), 202

Celery Concept

text
Flask request → Celery broker (Redis/RabbitMQ) → Worker process → result backend

Install:

bash
pip install celery redis

app/celery_app.py:

app/tasks.py:

python
from app.celery_app import celery
 
@celery.task
def send_welcome_email(user_id):
    user = db.session.get(User, user_id)
    mail.send(Message("Welcome", recipients=[user.email], body="Hello!"))

Run worker:

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

Trigger from view:

python
from app.tasks import send_welcome_email
 
send_welcome_email.delay(user.id)

Code explanation:

  • delay() publishes message to broker—returns immediately
  • Worker runs in separate process with Flask app context for DB access

Tip

Start With a Queue, Not Threads

threading in Flask does not survive process restarts—Celery/RQ scale with workers.

Lighter Alternative: RQ

For small projects:

bash
pip install rq
rq worker --with-scheduler

Same Redis broker idea—less configuration than Celery.

Idempotent Tasks

Retries may run twice—design tasks to tolerate duplicate execution (use unique job IDs, upsert DB rows).

FAQ

Cache shows stale data?

Lower TTL or invalidate on write; user-specific data needs user-specific keys.

Redis connection refused?

Start Redis: docker run -d -p 6379:6379 redis:7

Celery task cannot access db?

Wrap task body in app.app_context() via ContextTask.

Where to run Celery worker?

Separate systemd service or Docker container—not inside Gunicorn worker.

SimpleCache in production?

In-memory only—use Redis when multiple Gunicorn workers need shared cache.

Email in request thread?

Acceptable for single welcome email in dev—queue in production.

Skip this chapter?

Yes until you hit performance or async email/report requirements.