在Tornado中使用Future对象执行文件并返回状态的实现疑问

Hey Austin, let's break down what's going wrong with your current code and fix it step by step to meet your requirement—returning a "Pending" status immediately while running your conversion script in the background, then updating MongoDB once it's done.

First, Let's Spot the Issues in Your Current Code #

• Your _work method creates a Future but doesn't wait for it to resolve before returning "Pending". While the script might run, you have no logic to track when it finishes or update MongoDB afterward.
• The MongoDB queries use synchronous pymongo, which blocks Tornado's IOLoop—this defeats the purpose of using Tornado for async handling, as other requests will hang while these queries run.
• There's no way for users to check if the task completed after getting the initial "Pending" response.

Step-by-Step Solution #

First, switch to Motor (Tornado's async MongoDB driver) instead of synchronous pymongo to avoid blocking the IOLoop. Install it with:

pip install motor

1. Setup Async MongoDB Client

Initialize the Motor client when your app starts:

from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler
from tornado import gen
from tornado.subprocess import Subprocess
from motor.motor_tornado import MotorClient
import datetime

# Global async MongoDB client
client = MotorClient("mongodb://localhost:27017")
db = client.business

2. Async Task Handler (Returns "Pending" Immediately)

This handler starts the background script, saves a pending task to MongoDB, and sends the "Pending" response right away:

class FileAsyncHandler(RequestHandler):
    @gen.coroutine
    def get(self):
        # 1. Create a task record with Pending status in MongoDB
        task_result = yield db.tasks.insert_one({
            "status": "Pending",
            "created_at": datetime.datetime.utcnow()
        })
        task_id = str(task_result.inserted_id)

        # 2. Run the script in the background without blocking the request
        IOLoop.current().add_callback(self._execute_script, task_id)

        # 3. Return Pending status immediately
        self.write({"status": "Pending", "task_id": task_id})
        self.finish()

    @gen.coroutine
    def _execute_script(self, task_id):
        cmd = ['python3', '/home/Austin/Desktop/PythonCodes/magician.py']
        try:
            # Launch the script asynchronously
            p = Subprocess(cmd, stdout=Subprocess.STREAM, stderr=Subprocess.STREAM)
            # Wait for the script to finish (non-blocking)
            yield p.wait_for_exit()

            # 4. Update task status to Completed once done
            yield db.tasks.update_one(
                {"_id": task_id},
                {"$set": {
                    "status": "Completed",
                    "completed_at": datetime.datetime.utcnow()
                }}
            )
            print(f"Task {task_id} completed successfully")
        except Exception as e:
            # Handle script failures by updating status to Failed
            yield db.tasks.update_one(
                {"_id": task_id},
                {"$set": {
                    "status": "Failed",
                    "error": str(e),
                    "completed_at": datetime.datetime.utcnow()
                }}
            )
            print(f"Task {task_id} failed: {str(e)}")

3. Task Status Query Handler

Add a handler so users can check their task's status using the task_id they received:

class TaskStatusHandler(RequestHandler):
    @gen.coroutine
    def get(self, task_id):
        task = yield db.tasks.find_one({"_id": task_id})
        if not task:
            self.write({"error": "Task not found"})
            self.set_status(404)
        else:
            self.write({
                "task_id": task_id,
                "status": task["status"],
                "created_at": task.get("created_at"),
                "completed_at": task.get("completed_at"),
                "error": task.get("error")
            })
        self.finish()

4. Configure Your Tornado App

Wire up the handlers in your app:

def make_app():
    return Application([
        (r"/start-task", FileAsyncHandler),
        (r"/task-status/([^/]+)", TaskStatusHandler),
    ])

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    IOLoop.current().start()

Key Improvements Explained #

• Async MongoDB Operations: Motor ensures all database calls are non-blocking, so Tornado can handle other requests while waiting for DB actions.
• Background Task Execution: IOLoop.current().add_callback lets the script run in the background without holding up the initial request—this is why we can return "Pending" immediately.
• Task Tracking: The tasks collection keeps a record of each job's status, so users can poll /task-status/{task_id} to check if it's completed.
• Proper Error Handling: We catch script failures and update the task status accordingly, so users know if something went wrong.

Quick Notes #

• Ensure your magician.py script doesn't require interactive input (the subprocess runs without a terminal).
• If you need to capture the script's output, you can read from p.stdout and p.stderr (we included streams in the Subprocess setup for this).
• For production, add rate limiting or authentication to the endpoints, and schedule cleanup for old task records.

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