Flask：POST函数后台任务实现与跨类访问及水泵自动化控制问询

Hey there! Let's tackle your two Flask and automation requirements with practical, actionable code examples. I'll break this down into two clear sections so you can follow along easily.

To handle background tasks triggered by POST requests and make them accessible to other classes, we'll use Python's built-in threading module for lightweight task management, paired with a custom TaskManager class to keep track of all tasks. This approach avoids overcomplicating things with heavy task queues (though I'll mention a production-grade alternative later).

Step 1: Create a Task Manager #

This class will act as a central hub to start, track, and stop background tasks—making it easy for other classes to access task data.

from flask import Flask, request, jsonify
import threading
import time
from typing import Dict, Optional

app = Flask(__name__)

class TaskManager:
    def __init__(self):
        self.tasks: Dict[str, threading.Thread] = {}
        self.task_status: Dict[str, str] = {}

    def add_task(self, task_id: str, target, args=()):
        # Daemon thread ensures tasks exit when the Flask app stops
        thread = threading.Thread(target=target, args=args, daemon=True)
        self.tasks[task_id] = thread
        self.task_status[task_id] = "running"
        thread.start()

    def get_task(self, task_id: str) -> Optional[threading.Thread]:
        return self.tasks.get(task_id)

    def mark_task_for_stop(self, task_id: str):
        # Threads can't be force-stopped safely—use a status flag instead
        if self.task_status.get(task_id) == "running":
            self.task_status[task_id] = "stopping"

# Initialize a global task manager instance
task_manager = TaskManager()

Step 2: Trigger Tasks via POST #

Create a POST endpoint to start background tasks, and a sample task function that respects the stop flag:

def sample_background_task(task_id, duration):
    print(f"Task {task_id} started (running for {duration}s)")
    for i in range(duration):
        # Check if we need to exit early
        if task_manager.task_status.get(task_id) == "stopping":
            print(f"Task {task_id} stopped early")
            task_manager.task_status[task_id] = "stopped"
            return
        time.sleep(1)
    task_manager.task_status[task_id] = "completed"
    print(f"Task {task_id} finished")

@app.route('/start-task', methods=['POST'])
def start_task():
    data = request.json
    task_id = data.get('task_id')
    duration = data.get('duration', 10)

    if task_id in task_manager.tasks and task_manager.task_status[task_id] == "running":
        return jsonify({"error": "Task is already running"}), 400

    task_manager.add_task(task_id, sample_background_task, args=(task_id, duration))
    return jsonify({"message": f"Task {task_id} started in background"}), 200

Step 3: Access Tasks from Another Class #

Use the global task_manager instance to let other classes interact with tasks:

class TaskMonitor:
    def get_task_status(self, task_id):
        return task_manager.task_status.get(task_id, "Task not found")

# Test endpoint to demonstrate cross-class access
@app.route('/check-task/<task_id>', methods=['GET'])
def check_task(task_id):
    monitor = TaskMonitor()
    status = monitor.get_task_status(task_id)
    return jsonify({"task_id": task_id, "status": status})

Production-Grade Alternative #

For larger apps needing task persistence, retries, or distributed processing, use Celery with a message broker like Redis or RabbitMQ. It's more complex but scales better.

Now let's build the pump automation logic: a background task that continuously checks soil moisture, triggers the pump when thresholds are exceeded, and lets users stop the process anytime.

Step 1: Simulate Sensor & Pump Logic #

Replace these with your actual hardware control code (e.g., GPIO for Raspberry Pi):

def read_soil_moisture():
    # Replace with real sensor reading logic (e.g., ADC input)
    import random
    return random.randint(20, 80)  # Simulate moisture percentage (20-80%)

def control_water_pump(turn_on: bool):
    # Replace with real pump control code (e.g., GPIO.output(pin, turn_on))
    if turn_on:
        print("✅ Pump activated: Watering plants...")
    else:
        print("❌ Pump deactivated")

Step 2: Background Monitoring Task #

This task runs in the background, checks moisture levels, and controls the pump based on the user's threshold:

def pump_monitoring_task(task_id, threshold):
    print(f"🔍 Starting moisture monitoring (threshold: {threshold}%)")
    while task_manager.task_status.get(task_id) == "running":
        current_moisture = read_soil_moisture()
        print(f"Current moisture: {current_moisture}%")

        if current_moisture > threshold:
            control_water_pump(True)
            time.sleep(5)  # Keep pump on for 5 seconds
            control_water_pump(False)
        else:
            print("Moisture is below threshold—no watering needed")

        time.sleep(10)  # Wait 10 seconds before next check

    # Ensure pump is off when task stops
    control_water_pump(False)
    task_manager.task_status[task_id] = "stopped"
    print("🛑 Moisture monitoring terminated")

Step 3: User Endpoints to Start/Stop Automation #

Create POST endpoints for users to start the automation with a threshold and stop it anytime:

# Use a fixed task ID since we only need one active pump control process
PUMP_TASK_ID = "pump_automation"

@app.route('/start-pump-control', methods=['POST'])
def start_pump_control():
    data = request.json
    threshold = data.get('threshold')

    # Validate threshold input
    if not isinstance(threshold, int) or not (0 <= threshold <= 100):
        return jsonify({"error": "Invalid threshold—must be an integer between 0 and 100"}), 400

    if task_manager.task_status.get(PUMP_TASK_ID) == "running":
        return jsonify({"error": "Pump automation is already active"}), 400

    task_manager.add_task(PUMP_TASK_ID, pump_monitoring_task, args=(PUMP_TASK_ID, threshold))
    return jsonify({"message": f"Pump automation started (threshold: {threshold}%)"}), 200

@app.route('/stop-pump-control', methods=['POST'])
def stop_pump_control():
    if task_manager.task_status.get(PUMP_TASK_ID) != "running":
        return jsonify({"error": "Pump automation is not running"}), 400

    task_manager.mark_task_for_stop(PUMP_TASK_ID)
    return jsonify({"message": "Pump automation is stopping..."}), 200

Key Notes #

• Safe Thread Termination: We never force-kill threads—instead, we use a status flag to let the task exit gracefully. This prevents resource leaks.
• Hardware Safety: Always ensure the pump is turned off when the task stops to avoid flooding.
• Sensor Calibration: Adjust the read_soil_moisture function to match your sensor's output range.

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