Start a Flask Application in Separate Thread

Start a flask application in separate thread

You're running Flask in debug mode, which enables the reloader (reloads the Flask server when your code changes).

Flask can run just fine in a separate thread, but the reloader expects to run in the main thread.

---

To solve your issue, you should either disable debug (app.debug = False), or disable the reloader (app.use_reloader=False).

Those can also be passed as arguments to app.run: app.run(debug=True, use_reloader=False).

How to make a post request to a flask server running on a separate thread

To isolate an app for every test function, you should use @pytest.fixture decorator.

First of all, you should read Testing Flask Applications from the official docs.

Start to define, a new fixture for your app in your conftest.py, with a function scope (default), other scopes.

from myapp import create_app

@pytest.fixture
def app():
    app = create_app()
    return app

Then, inside your test function, you can call your app fixture.

def test_my_function(app):
    # Here you have a fresh app object.
    pass

Here starts another part. You need a test_client to make requests. Note that if you are testing for assertions or exceptions in your application code, you must set app.testing = True in order for the exceptions to propagate to the test client.

An easier way to do that is to use the pytest-flask which is going to create you the other fixtures that you need like client, config, and others...

Example with pytest-flask:

def test_my_function(client):
    rv = client.get('/')
    assert b'No entries here so far' in rv.data

If you take a look at pytest-flask/fixture.py on github, you can see that the client fixture is just depending on an app fixture, and returns a text_client from it.

You can still try it to do it with thread, but this way is much simpler. You do not have to care about how to end the thread. You might have another problem with the debug mode that starts the main thread to control the real app.

In the end, for each test function that calls the client fixture, you will have a new fresh isolated app and this is what you wanted.

How to start another thread in request handling thread with Flask?

The code works for me

def test_multi_threading_query():
    # module which i create Flask app instance
    from app.main import app
    # module which i create sqlalchemhy instance
    from app.model.db import db, Post
    with app.app_context():
        posts = Post.query.all()
        p = posts[0]
        p.foo = 1
        db.session.add(p)
        db.session.commit()
        print(p)

@api.route('/test')
def test_view():
    from threading import Thread
    t = Thread(target=test_multi_threading_query)
    t.start()
    return ''

# main.py
app = Flask(__main__)

#db.py
db = SQLAlchemy()

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(db.Integer)

https://flask.palletsprojects.com/en/1.1.x/appcontext/

How to start flask api server in separate thread in Python

I have resolved the issue by starting a separate thread to run API server:

run = True

def start_api_server():
    while run:
        start_local_server()
        
    time.sleep(1)
    print("SERVER HAS STOPPED")

@pyqtSlot()
def on_click_start_btn(self):
    Thread(target=start_api_server).start()

Properly terminate flask web app running in a thread

Dont join child thread. Use setDaemon instead:

from flask import Flask
import time
import threading

def thread_webAPP():
    app = Flask(__name__)

    @app.route("/")
    def nothing():
        return "Hello World!"

    app.run(debug=True, use_reloader=False)

t_webApp = threading.Thread(name='Web App', target=thread_webAPP)
t_webApp.setDaemon(True)
t_webApp.start()

try:
    while True:
        time.sleep(1)

except KeyboardInterrupt:
    print("exiting")
    exit(0)

daemon for a child thread means that the main thread won't wait till this daemon child thread is finished its job if you're trying to stop the main thread. In this case all child threads will be joined automatically and the main thread will be successfully stopped immediately.

More info is here.

Flask using thread to run code after/alognside return

it does work, but since in your Hello function you return 'Hello' after the thread start, the return redirect("/goodbye") never makes it to the browser. But the processing in the background works, you can add some print statements in the long_running_task code and see the flask console for yourself:

@app.route("/hello")
def Hello():
    thread = threading.Thread(target=long_running_task)
    thread.start()
    return 'hello'

def long_running_task():
    print("long task started")
    time.sleep(5)
    print("long task ended")
    return redirect("/goodbye")

Can I serve multiple clients using just Flask app.run() as standalone?

flask.Flask.run accepts additional keyword arguments (**options) that it forwards to werkzeug.serving.run_simple - two of those arguments are threaded (a boolean) and processes (which you can set to a number greater than one to have werkzeug spawn more than one process to handle requests).

threaded defaults to True as of Flask 1.0, so for the latest versions of Flask, the default development server will be able to serve multiple clients simultaneously by default. For older versions of Flask, you can explicitly pass threaded=True to enable this behaviour.

For example, you can do

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

to handle multiple clients using threads in a way compatible with old Flask versions, or

if __name__ == '__main__':
    app.run(threaded=False, processes=3)

to tell Werkzeug to spawn three processes to handle incoming requests, or just

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

to handle multiple clients using threads if you know that you will be using Flask 1.0 or later.

That being said, Werkzeug's serving.run_simple wraps the standard library's wsgiref package - and that package contains a reference implementation of WSGI, not a production-ready web server. If you are going to use Flask in production (assuming that "production" is not a low-traffic internal application with no more than 10 concurrent users) make sure to stand it up behind a real web server (see the section of Flask's docs entitled Deployment Options for some suggested methods).

Running a Flask server parallel to main app

May be its better to run your game in s different thread?

import threading
import time
from flask import Flask, render_template

class myGame(threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.board = 1

    def run(self):
        pass

app = Flask(__name__)
game = myGame()

@app.route('/get_updates') 
def get_updates(): 
    return flask.jsonify(game.board)

if __name__ == "__main__":
    game.start()
    app.run(port=81, host='0.0.0.0', debug=False, use_reloader=False)

---

Related Topics