Skip to content

Instantly share code, notes, and snippets.

@gmolveau
Last active March 6, 2023 07:27
Show Gist options
  • Star 8 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save gmolveau/10c80785d499c2e2d4c447343a624664 to your computer and use it in GitHub Desktop.
Save gmolveau/10c80785d499c2e2d4c447343a624664 to your computer and use it in GitHub Desktop.
run a flask application in a separate thread
# please use python3 and flask > 1.0
# run this with : python3 flask_separate_thread.py
# https://web.archive.org/web/20210329135424/https://pymotw.com/3/threading/
# https://web.archive.org/web/20210329135553/https://eli.thegreenplace.net/2011/08/22/how-not-to-set-a-timeout-on-a-computation-in-python
import time
import threading
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return 'Hello, World!'
def web():
app.run(debug=True, use_reloader=False, host='0.0.0.0', port=5000)
def usb(arg1, arg2):
print(f"this is how to pass arguments to a thread function - {arg1} - {arg2}")
while True:
print("waiting for USB")
time.sleep(2)
if __name__ == '__main__':
threading.Thread(target=web, daemon=True).start()
threading.Thread(target=usb, daemon=True).start()
while True:
time.sleep(1)
@grufocom
Copy link

hi, do you know how to call the flask-hello_world from inside of the usb-thread?

@gmolveau
Copy link
Author

@grufocom I think you should extract the logic from hello_world and put it in another function. You can then call this function both in the usb thread and the hello_world route.

@vvalenzuela96
Copy link

How I can deploy this on production if flask docs say don't use run() in a production setting?

@HansG89
Copy link

HansG89 commented Feb 9, 2023

How I can deploy this on production if flask docs say don't use run() in a production setting?

By using a WSGI compliant server like Gunicorn (see Flask docs)

@vvalenzuela96
Copy link

How I can deploy this on production if flask docs say don't use run() in a production setting?

By using a WSGI compliant server like Gunicorn (see Flask docs)

Thanks for your reply. I discovered a way to make a threaded Flask app

@ASRodrigo1
Copy link

How I can deploy this on production if flask docs say don't use run() in a production setting?

By using a WSGI compliant server like Gunicorn (see Flask docs)

Thanks for your reply. I discovered a way to make a threaded Flask app

Can you share it please?

@vvalenzuela96
Copy link

vvalenzuela96 commented Feb 27, 2023

How I can deploy this on production if flask docs say don't use run() in a production setting?

By using a WSGI compliant server like Gunicorn (see Flask docs)

Thanks for your reply. I discovered a way to make a threaded Flask app

Can you share it please?

Well, I separated my code into files, but I will explain each one. I work with Blueprints for my route functions.
Work with apache-XAMPP as webserver and Waitress as Python WSGI server.

First, I used a project structure that used this tutorial (but it is in Spanish). My project structure is:

project_dir/
|__ entry_point.py
|__ app/
    |__ __init__.py
    |__ updater.py
    |__ some_module/
        |__ __init__.py
        |__ routes.py
        |__ models.py
        |__ templates
            |__ some_module.html

The entry_point.py has the instance of the Flask application that calls the generate_app() function on app/__init__.py:

# entry_point.py

import os
from app import generate_app


settings_module = os.getenv('APP_SETTINGS_MODULE')
application = generate_app(settings_module=settings_module)

The app/__init__.py has the function that generates the Flask application. Any other thread should be initiated here:

# app/__init__.py

import threading

from flask import Flask
from werkzeug.middleware.proxy_fix import ProxyFix
from flask_login import LoginManager
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from werkzeug.middleware.proxy_fix import ProxyFix
...

from app.updater import update # The function that runs in another thread


login_manager = LoginManager()
db = SQLAlchemy()
migrate = Migrate()
...

def generate_app(settings_module='config.DevelopmentConfig'):
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_object(settings_module)

    # WSGI PROXY CONFIGURATION
    app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)

    # MODULES INITIALIZATION
    login_manager.init_app(app)
    db.init_app(app)
    migrate.init_app(app, db)
    ...

    register_blueprints(app)
    register_error_handlers(app)
    
    # THREAD INITIALIZATION 
    threading.Thread(target=update, args=(app,), daemon=True).start()
    ...

    return app

In the app/updater.py file, should be written the function that runs in the other thread, that it will be called by generate_app() in app/__init__.py:

# app/updater.py

def update(application):
    with application.app_context():
        print('ANOTHER THREAD')
        ...

Finally, it will be started with: flask run command or waitress-serve --listen=127.0.0.1:5000 entry_point:application command. The Waitress command is for production.

Note: The debug mode may duplicate the Flask app

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment