OpenFactoryFlaskApp#

class openfactory.apps.ofa_flask_app.OpenFactoryFlaskApp(ksqlClient, bootstrap_servers=None, asset_router_url=None, loglevel='INFO', log_http_requests=True, 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.

Execution model:

run() starts both the OpenFactory runtime and an embedded Flask server.
Flask executes in a dedicated background thread.
OpenFactory application logic executes asynchronously in the asyncio event loop.
The HTTP server listens on 0.0.0.0.
The listening port is taken from PORT or defaults to 4000.
Exceptions from the Flask thread are propagated to the asyncio runtime.

Customization:

Override create_flask_app() to customize the Flask application.
Override configure_routes() to register routes or Blueprints.
Override async_main_loop() to implement asynchronous background tasks.
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 = super().create_flask_app()
        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.
HTTP access logging can be enabled or disabled using the log_http_requests constructor parameter.
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

The default implementation creates the Flask application using the OpenFactory module instead of the subclass module. This ensures Flask can correctly locate the framework’s bundled templates and static resources. Subclasses should normally customize the returned application rather than creating a new Flask instance unless a completely custom application object is required.

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

Example

class DemoFlaskApp(OpenFactoryFlaskApp):

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

Note

The returned Flask application is automatically exposed through self.app
create_flask_app() is invoked before configure_routes(), allowing subclasses to configure the application before routes and Blueprints are registered.

run()[source]#

Start the application using a synchronous entry point.

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