OpenFactoryFlaskApp#

class openfactory.apps.ofa_flask_app.OpenFactoryFlaskApp(ksqlClient, bootstrap_servers=None, asset_router_url=None, loglevel='INFO', test_mode=False)[source]#

Bases: OpenFactoryApp

OpenFactory application with an embedded Flask web interface.

Extends OpenFactoryApp by attaching a flask.Flask application to the OpenFactory runtime. This allows exposing HTTP endpoints alongside the standard OpenFactory asset, attribute, and method mechanisms.

The Flask application is available via the app attribute and can be used exactly as in a standard Flask project.

Runtime behavior:

Calling run() starts both the OpenFactory application and an embedded HTTP server.
The Flask server runs in a dedicated background thread using Werkzeug.
The OpenFactory logic runs asynchronously inside the asyncio event loop.
The server listens on all network interfaces (0.0.0.0).
The listening port is defined by the PORT environment variable, or defaults to 4000 if unset.
Unhandled exceptions in the Flask thread are propagated to the main asyncio runtime.

Architecture:

Flask runs in a dedicated thread
OpenFactory logic runs in asyncio
User applications implement async_main_loop()
The synchronous OpenFactoryApp.main_loop() is intentionally not supported

app#

Flask application instance attached to the OpenFactory app.

Type:: flask.Flask

Usage Example (inline routes)

import os
import asyncio
from flask import current_app
from openfactory.apps import OpenFactoryFlaskApp, EventAttribute, ofa_method
from openfactory.kafka import KSQLDBClient

class DemoFlaskApp(OpenFactoryFlaskApp):

    status = EventAttribute(value="idle", tag="App.Status")

    def configure_routes(self):

        @self.app.route("/")
        def root():
            ofa_app = current_app.ofa_app
            return {
                "status": ofa_app.status.value
            }

    @ofa_method(description="Move axis")
    def move_axis(self, x: float, y: float):
        self.logger.info(f"Move to {x},{y}")

    async def async_main_loop(self):
        while True:
            await asyncio.sleep(5)
            self.logger.info("Background task running")

app = DemoFlaskApp(
    ksqlClient=KSQLDBClient(os.getenv("KSQLDB_URL", "http://localhost:8088")),
    bootstrap_servers=os.getenv("KAFKA_BROKER", "localhost:9092"),
)

app.run()

For larger applications, routes can be split into modules and registered using Blueprints:

Using Blueprints (recommended for larger applications)

# main.py
import os
import asyncio
from openfactory.apps import OpenFactoryFlaskApp
from openfactory.kafka import KSQLDBClient
from routes.root import root_bp

class DemoFlaskApp(OpenFactoryFlaskApp):

    def configure_routes(self):
        self.app.register_blueprint(root_bp)

    async def async_main_loop(self):
        while True:
            await asyncio.sleep(5)

app = DemoFlaskApp(
    ksqlClient=KSQLDBClient(os.getenv("KSQLDB_URL", "http://localhost:8088")),
    bootstrap_servers=os.getenv("KAFKA_BROKER", "localhost:9092"),
)

app.run()

# routes/root.py
from flask import Blueprint, current_app

root_bp = Blueprint("root", __name__)

@root_bp.route("/")
def root():
    ofa_app = current_app.ofa_app
    return {
        "availability": ofa_app.avail.value
    }

For applications requiring additional Flask configuration or extension setup, the Flask application instance can also be customized using the application factory pattern:

Using a Flask application factory

import os
import asyncio
from flask import Flask
from openfactory.apps import OpenFactoryFlaskApp
from openfactory.kafka import KSQLDBClient

class DemoFlaskApp(OpenFactoryFlaskApp):

    def create_flask_app(self):
        app = Flask(__name__)
        app.config["SECRET_KEY"] = "my-secret-key"
        return app

    def configure_routes(self):

        @self.app.route("/")
        def root():
            return {
                "message": "Hello Flask"
            }

app = DemoFlaskApp(
    ksqlClient=KSQLDBClient(os.getenv("KSQLDB_URL", "http://localhost:8088")),
    bootstrap_servers=os.getenv("KAFKA_BROKER", "localhost:9092"),
)

app.run()

Note

The Flask application is accessible via app and behaves like a standard Flask instance.
Routes can be registered directly using flask.Flask.route() or through Flask Blueprints (flask.Blueprint).
Only asynchronous execution is supported. Subclasses should implement async_main_loop() for background tasks.
The synchronous OpenFactoryApp.main_loop() is not supported in this class.
OpenFactory features such as attributes, methods, and asset communication remain unchanged.
When deployed on the OpenFactory platform, the PORT environment variable is set automatically by the deployment tool.
OpenFactory automatically enables Werkzeug ProxyFix to support reverse-proxy deployments through the OpenFactory Traefik routing gateway, including localhost path-prefix routing in development mode.
When deployed on OpenFactory, APPLICATION_ROOT is automatically configured from the OPENFACTORY_ROOT_PATH environment variable injected by the OpenFactory deployment tool, unless explicitly defined by the child class in create_flask_app(). This ensures correct Flask behavior for redirects, URL generation, Blueprint routing, and session cookie handling under path-prefix deployments.

See also

OpenFactoryApp for full initialization details and environment variable handling.

async async_main_loop()[source]#

Default asynchronous main loop.

Keeps the application alive when no custom loop is provided.

Subclasses can override this method to implement background logic.

Example

async def async_main_loop(self):
    while True:
        await asyncio.sleep(5)
        self.logger.info("Background task running")

Return type:: None

async async_run()[source]#

Asynchronous entry point of the application.

Starts both the Flask server and the OpenFactory logic concurrently.

This method:

Displays the welcome banner
Sets the application availability
Starts the Flask server thread
Runs the OpenFactory async logic
Propagates exceptions from the Flask thread
Gracefully shuts down the application

Raises:: Exception – Any unhandled exception during execution is logged and triggers OpenFactoryApp.app_event_loop_stopped().

configure_routes()[source]#

Configure HTTP routes for the Flask application.

This method can be overridden by subclasses to define routes directly using app.

Example

def configure_routes(self):

    @self.app.route("/")
    def root():
        return {"status": "ok"}

Note

For larger applications, prefer using Flask Blueprints instead of defining all routes inline.

create_flask_app()[source]#

Create and configure the Flask application instance.

This method is called during initialization to create the flask.Flask application attached to the OpenFactory runtime.

Subclasses may override this method to customize the Flask application before routes are registered via configure_routes().

Return type:: Flask

Typical use cases include:

Loading Flask configuration
Initializing Flask extensions
Enabling middleware
Customizing template or static paths
Applying testing configuration

Returns:: flask.Flask – Configured Flask application instance.

Example

class DemoFlaskApp(OpenFactoryFlaskApp):

    def create_flask_app(self):
        app = Flask(__name__)
        app.config["SECRET_KEY"] = "my-secret"
        return app

Note

The returned Flask application is automatically exposed through:

self.app

allowing subclasses to configure routes, Blueprints, extensions, and Flask application settings

run()[source]#

Start the application using a synchronous entry point.

This method runs async_run() inside an asyncio event loop.