Home - News ( United States | United Kingdom | Italy | Germany ) - Football scores

Showing content from https://stackoverflow.com/questions/79003376/flask-blueprint-with-socketio-configuration below:

socket.io - Flask Blueprint with socketio configuration

My solution was instead of trying to start_background_task inside the blueprint on its initiation, I now start the task inside create_app(), giving up on the idea on initializing everything inside the blueprint.

The main difference from before is that now all of my configuration is in a separate file. By importing my threaded function into create_app() and then passing socketio back to it as a parameter, I'm avoiding all on the circular imports and initiation issues.

This also makes it pretty much pointless to create a blueprint for the sockets as just a regular .py works.

Thanks to everyone that helped!! 💪

create_app() -

from flask import Flask
from flask_socketio import SocketIO
import os

from flask_app.sockets.background_tasks import emit_logs

socketio = SocketIO()

def create_app(debug=False):
    app = Flask(__name__)

    app.debug = debug
    app.config['SECRET_KEY'] = os.urandom(24)

    # Initialize SocketIO.
    socketio.init_app(app, cors_allowed_origins="*")    

    # Start background tasks.
    socketio.start_background_task(emit_logs, socketio)

    return app

background_tasks.py -

# Threaded function to emit logs to client
# Now takes a socketio object!!!
def emit_logs(socketio: SocketIO):
    print('Emitting logs to FE')
    db_handler = database_handler()
    db_handler.connect()

    try:
        while True:
            recent_logs = [1, 2, 3]

            socketio.emit('new_log', {'data': recent_logs})

            socketio.sleep(5)  # Fetch and emit every 1 seconds
    except Exception as e:
        print(f"Error fetching/emitting logs: {str(e)}")
    finally:
        db_handler.disconnect()

With this configuration, using socketio normally (by importing it from create_app), causes a circular import.

To prevent this, in sockets/events I defined an 'initialize_socketio()' function that waits until create_app() passes the socketio instance to it as a parameter, and then uses it to configure the events:

from flask_socketio import SocketIO

# This is ran in create_app() to initialize socketio's events.
def initialize_socketio(socketio: SocketIO):

    @socketio.on("test_socket")
    def test_socket():
        print("Got a message!")

And then call it in create_app() -

from flask import Flask
from flask_socketio import SocketIO

from flask_app.blueprints.sockets.events import initialize_socketio

socketio = SocketIO()

def create_app(debug=False):
    # Initializing Flask
    app = Flask(__name__)

    # Initialize SocketIO.
    socketio.init_app(app, cors_allowed_origins="*")     
    initialize_socketio(socketio)

Hope that helps!!

RetroSearch is an open source project built by @garambo | Open a GitHub Issue

Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo

HTML: 3.2 | Encoding: UTF-8 | Version: 0.7.4