如何查询循环中Python类的数据？Flask后台运行循环并获取值实现

Question 1: Accessing the current Attribute #

In your original code, when you create obj = L(), the __init__ method immediately calls self.loop()—which runs an infinite while loop that blocks the main thread forever. You’ll never get to a point where you can interact with obj to access current because the script is stuck in the loop.

If you modify the code to run the loop in a separate background thread (as we’ll cover in Question 2), accessing current is trivial: just use obj.current directly. For example, you could print it, return it in a web endpoint, or send it as a Telegram message.

Question 2: Running the Loop in the Background (Flask & Telegram Examples) #

Absolutely! The core solution here is to run the infinite loop in a separate thread so it doesn’t block your main application (whether that’s a Flask web app or a Telegram bot). Here’s how to implement both:

Flask Implementation #

This setup lets you start the loop via a /start endpoint and fetch the current value with /get:

import threading
import time
from flask import Flask

class L:
    def __init__(self):
        self.data = []
        self.current = 0
        self.loop_thread = None
        self.lock = threading.Lock()  # Prevents race conditions

    def loop(self):
        while True:
            with self.lock:
                self.data.append(self.current)
                self.current += 1
            time.sleep(2)

    def get_current_value(self):
        with self.lock:
            return self.current

# Initialize Flask app and our loop class instance
app = Flask(__name__)
loop_obj = L()

@app.route('/start')
def start_loop():
    if not loop_obj.loop_thread or not loop_obj.loop_thread.is_alive():
        # Start loop in a daemon thread (shuts down when main app exits)
        loop_obj.loop_thread = threading.Thread(target=loop_obj.loop, daemon=True)
        loop_obj.loop_thread.start()
        return "Background loop started successfully!"
    return "Loop is already running."

@app.route('/get')
def fetch_current():
    return f"Current value: {loop_obj.get_current_value()}"

if __name__ == '__main__':
    app.run(debug=True)

Key Details:

• We added a threading.Lock() to safely read/write the current attribute (prevents race conditions between the background thread and main app thread).
• The daemon thread ensures the loop stops automatically when you shut down the Flask app.
• The /start endpoint checks if the loop is already running to avoid duplicate threads.

Telegram Bot Implementation #

For a Telegram bot using python-telegram-bot, the approach is nearly identical—we just hook into command handlers instead of Flask routes:

import threading
import time
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

class L:
    def __init__(self):
        self.data = []
        self.current = 0
        self.loop_thread = None
        self.lock = threading.Lock()

    def loop(self):
        while True:
            with self.lock:
                self.data.append(self.current)
                self.current += 1
            time.sleep(2)

    def get_current_value(self):
        with self.lock:
            return self.current

# Initialize loop instance
loop_obj = L()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not loop_obj.loop_thread or not loop_obj.loop_thread.is_alive():
        loop_obj.loop_thread = threading.Thread(target=loop_obj.loop, daemon=True)
        loop_obj.loop_thread.start()
        await update.message.reply_text("Loop started! Use /get to check the current value.")
    else:
        await update.message.reply_text("Loop is already running.")

async def get_current(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(f"Current value: {loop_obj.get_current_value()}")

if __name__ == '__main__':
    # Replace with your bot token from @BotFather
    app = ApplicationBuilder().token("YOUR_TELEGRAM_BOT_TOKEN").build()
    
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("get", get_current))
    
    app.run_polling()

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