基于Flask路由实现Python同步进程的启停控制

Got it, let's tackle this problem step by step. Since you're using Flask with Python 3.6 to build a tool that syncs data from sys1 to sys2 continuously, and need to control its start/stop via Flask routes, here's a practical implementation that avoids blocking your Flask app and handles proper HTTP responses:

Core Approach #

The key here is to run the sync process in a background thread so it doesn't block your Flask app's main thread (which handles incoming requests). We'll use a thread-safe event flag to control when the sync process stops, and expose /start and /stop routes to manage the process.

Full Code Example #

from flask import Flask, jsonify
import threading
import time
# Uncomment below if you need requests for REST calls (Python 3.6 supports requests)
# import requests

app = Flask(__name__)

# Thread-safe flag to control the sync lifecycle
stop_sync_event = threading.Event()
sync_thread = None

def sync_data_from_sys1_to_sys2():
    """Continuous sync loop: fetch data from sys1, push to sys2"""
    while not stop_sync_event.is_set():
        try:
            # Step 1: Fetch data from sys1 (replace with actual REST call)
            sys1_data = get_data_from_sys1()
            if sys1_data:
                # Step 2: Send data to sys2 (replace with actual REST call)
                send_data_to_sys2(sys1_data)
                print("Data synced successfully from sys1 to sys2")
            else:
                print("No new data available for sync")
        except Exception as e:
            print(f"Sync failed with error: {str(e)}")
        
        # Add a delay between sync cycles (adjust based on your needs)
        time.sleep(10)

def get_data_from_sys1():
    """Mock function: replace with your actual GET request to sys1's API"""
    # Example: return requests.get("https://sys1-api.example.com/latest-data").json()
    return {"record_id": 123, "content": "sample data from sys1"}

def send_data_to_sys2(data):
    """Mock function: replace with your actual POST/PUT request to sys2's API"""
    # Example: requests.post("https://sys2-api.example.com/ingest", json=data)
    pass

@app.route('/start', methods=['POST'])
def start_sync():
    global sync_thread
    if sync_thread and sync_thread.is_alive():
        # Sync process is already running
        return jsonify({"message": "Sync process is already active"}), 409
    
    # Reset stop flag and launch the sync thread
    stop_sync_event.clear()
    sync_thread = threading.Thread(target=sync_data_from_sys1_to_sys2, daemon=True)
    sync_thread.start()
    
    return jsonify({"message": "Sync process started successfully"}), 200

@app.route('/stop', methods=['POST'])
def stop_sync():
    if not sync_thread or not sync_thread.is_alive():
        # No active sync process to stop
        return jsonify({"message": "No sync process is currently running"}), 400
    
    # Trigger stop signal and wait for thread to clean up (optional timeout)
    stop_sync_event.set()
    sync_thread.join(timeout=5)  # Wait up to 5 seconds for graceful exit
    
    return jsonify({"message": "Sync process stopped successfully"}), 200

if __name__ == '__main__':
    app.run(debug=True, use_reloader=False)
    # Note: use_reloader=False is critical in debug mode to avoid duplicate sync threads

Key Details Explained #

• Background Thread: Using threading.Thread ensures the sync logic runs independently without blocking Flask's request handling. The daemon=True flag makes sure the thread exits when the main Flask app shuts down.
• Thread-Safe Control: threading.Event is preferred over a plain global variable because it’s designed for safe cross-thread signaling. stop_sync_event.set() tells the sync loop to exit, while clear() resets it for a new session.
• HTTP Response Codes:
  • /start: Returns 200 OK when the sync starts, 409 Conflict if it’s already running.
  • /stop: Returns 200 OK when the sync stops, 400 Bad Request if there’s no active sync process.
• Python 3.6 Compatibility: All syntax here (f-strings, threading API) works seamlessly with Python 3.6.
• Debug Mode Note: Disable Flask’s auto-reloader (use_reloader=False) in debug mode to prevent duplicate sync threads from spawning.

Customization Tips #

• Replace the mock functions get_data_from_sys1() and send_data_to_sys2() with your actual REST calls (use the requests library for this).
• Adjust the time.sleep(10) interval to match your desired sync frequency.
• Add robust error handling (e.g., retry logic for failed API calls, file logging instead of print statements) based on your use case.

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