Flask应用：如何定时刷新全局变量？含数据库更新触发场景

Got it, let's tackle this problem step by step. You've got a Flask app where you're storing a list from a RESTDB API as a global items variable, and you need to refresh it both when you write data to the API and on a schedule. Here's how to do it properly:

1. Refactor Your Item Loading Logic First #

Instead of initializing items directly in __init__.py, wrap the RESTDB fetch logic in a reusable function. This makes it easy to trigger refreshes whenever you need to.

In your __init__.py:

from flask import Flask
import requests

app = Flask(__name__)
items = []  # Your global variable

def load_items_from_restdb():
    """Fetch the latest items from RESTDB and update the global variable."""
    global items
    try:
        # Replace with your actual RESTDB endpoint and auth if needed
        response = requests.get("https://your-restdb-instance.com/api/items", timeout=5)
        response.raise_for_status()  # Raise an error for HTTP status codes >=400
        items = response.json()
        app.logger.info("Successfully refreshed items from RESTDB")
    except requests.exceptions.RequestException as e:
        app.logger.error(f"Failed to refresh items: {str(e)}")
        # Keep the existing items instead of overwriting with empty/invalid data
        pass

# Load items once when the app starts
load_items_from_restdb()

2. Trigger Refresh After Writing to RESTDB #

When your view writes data to RESTDB, simply call load_items_from_restdb() right after the write operation to sync the global items variable.

In someview.py:

from project import app, load_items_from_restdb, items
import requests

@app.route('/add-item', methods=['POST'])
def add_item():
    # Example: Write a new item to RESTDB
    new_item = {"name": "New Item", "value": 42}
    try:
        response = requests.post(
            "https://your-restdb-instance.com/api/items",
            json=new_item,
            timeout=5
        )
        response.raise_for_status()
        # Refresh the global items immediately after a successful write
        load_items_from_restdb()
        return "Item added successfully! Items refreshed."
    except requests.exceptions.RequestException as e:
        return f"Failed to add item: {str(e)}", 500

3. Set Up Scheduled Auto-Refresh #

For periodic refreshes (even when no writes happen), use the APScheduler library—it's the standard tool for background tasks in Flask.

First, install it:

pip install apscheduler

Then update your __init__.py to add the scheduler:

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
from threading import Lock
import requests

app = Flask(__name__)
items = []
items_lock = Lock()  # For thread-safe access

def load_items_from_restdb():
    global items
    with items_lock:  # Prevent race conditions during refresh
        try:
            response = requests.get("https://your-restdb-instance.com/api/items", timeout=5)
            response.raise_for_status()
            items = response.json()
            app.logger.info("Scheduled items refresh completed")
        except requests.exceptions.RequestException as e:
            app.logger.error(f"Scheduled refresh failed: {str(e)}")
            pass

# Initial load
load_items_from_restdb()

# Configure scheduler (e.g., refresh every 5 minutes)
scheduler = BackgroundScheduler()
# Use 'seconds=30' for testing, adjust to your needs
scheduler.add_job(func=load_items_from_restdb, trigger="interval", minutes=5)
scheduler.start()

# Stop the scheduler when the app shuts down
@app.teardown_appcontext
def shutdown_scheduler(exception=None):
    if scheduler.running:
        scheduler.shutdown()

Important Pitfalls to Avoid #

• Multi-Worker Issues: If you're running Flask with a multi-worker server (like Gunicorn with --workers 3), each worker will have its own copy of items and its own scheduler. To fix this, consider using a shared cache (like Redis) instead of an in-memory global variable, or configure the scheduler to run only in the main worker.
• Thread Safety: The items_lock ensures that no request reads items while it's being updated, preventing race conditions. Always use locks when modifying global variables accessed by multiple threads.
• Error Resilience: The try/except blocks in load_items_from_restdb() prevent your app from breaking if RESTDB is unavailable—your app will keep using the last valid version of items.

内容的提问来源于stack exchange，提问作者caliph