try:
from flask import Flask
except ImportError as e:
raise ImportError(
"Flask is required for OpenFactoryFlaskApp. "
"Install with: pip install openfactory[flask]"
) from e
import os
import asyncio
import threading
from werkzeug.serving import make_server
from werkzeug.middleware.proxy_fix import ProxyFix
from werkzeug.serving import WSGIRequestHandler
from openfactory.apps import OpenFactoryApp
from openfactory.kafka import KSQLDBClient
class QuietWSGIRequestHandler(WSGIRequestHandler):
def log_request(self, code="-", size="-"):
pass
[docs]
class OpenFactoryFlaskApp(OpenFactoryApp):
"""
OpenFactory application with an embedded Flask web interface.
Extends :class:`OpenFactoryApp <openfactory.apps.ofaapp.OpenFactoryApp>` by attaching a
:class:`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 :attr:`app` attribute and
can be used exactly as in a standard Flask project.
Execution model:
- :meth:`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 :meth:`create_flask_app` to customize the Flask application.
- Override :meth:`configure_routes` to register routes or Blueprints.
- Override :meth:`async_main_loop` to implement asynchronous background tasks.
- The synchronous :meth:`OpenFactoryApp.main_loop() <openfactory.apps.ofaapp.OpenFactoryApp.main_loop>` is intentionally not supported.
Attributes:
app (flask.Flask): Flask application instance attached to the OpenFactory app.
.. admonition:: Usage Example (inline routes)
.. code-block:: python
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:
.. admonition:: Using Blueprints (recommended for larger applications)
.. code-block:: python
# 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()
.. code-block:: python
# 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:
.. admonition:: Using a Flask application factory
.. code-block:: python
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 :attr:`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 :meth:`flask.Flask.route` or through Flask Blueprints (:class:`flask.Blueprint`).
- Only asynchronous execution is supported. Subclasses should implement :meth:`async_main_loop`
for background tasks.
- The synchronous :meth:`OpenFactoryApp.main_loop() <openfactory.apps.ofaapp.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 :meth:`create_flask_app`.
This ensures correct Flask behavior for redirects, URL generation, Blueprint routing, and
session cookie handling under path-prefix deployments.
.. seealso::
- :class:`openfactory.apps.ofaapp.OpenFactoryApp`
- :class:`flask.Flask`
- `Flask documentation <https://flask.palletsprojects.com/>`_
"""
[docs]
def __init__(
self,
ksqlClient: KSQLDBClient,
bootstrap_servers: str | None = None,
asset_router_url: str | None = None,
loglevel: str = "INFO",
log_http_requests: bool = True,
test_mode: bool = False,
):
"""
Initialize the OpenFactory Flask application.
This constructor forwards all parameters to
:class:`OpenFactoryApp <openfactory.apps.ofaapp.OpenFactoryApp>`
and additionally creates a :class:`flask.Flask` instance
accessible via :attr:`app`.
The OpenFactory application instance is exposed inside Flask through:
.. code-block:: python
current_app.ofa_app
allowing Flask routes and Blueprints to access the OpenFactory runtime.
Args:
ksqlClient: KSQL client instance.
bootstrap_servers: Kafka bootstrap server address.
asset_router_url: Asset Router URL.
loglevel: Logging level (e.g., ``INFO``, ``DEBUG``).
log_http_requests: Enables logging of incoming HTTP requests handled by the embedded Flask server.
test_mode: Enables test mode (disables live Kafka/ksql interaction).
See also:
:class:`OpenFactoryApp <openfactory.apps.ofaapp.OpenFactoryApp>`
for full initialization details and environment variable handling.
"""
super().__init__(
ksqlClient=ksqlClient,
bootstrap_servers=bootstrap_servers,
asset_router_url=asset_router_url,
loglevel=loglevel,
test_mode=test_mode
)
self.log_http_requests = log_http_requests
# Flask application
self.app = self.create_flask_app()
# Trust reverse proxy forwarded headers.
#
# Required for correct URL generation and redirects when
# applications are exposed behind a path prefix through
# the OpenFactory Traefik development gateway:
#
# http://localhost/<app-name>/
#
# Traefik StripPrefix forwards the original external
# prefix via X-Forwarded-Prefix, which Werkzeug maps
# to SCRIPT_NAME through ProxyFix.
self.app.wsgi_app = ProxyFix(
self.app.wsgi_app,
x_prefix=1,
x_host=1,
x_proto=1,
)
# set root path if deployed on OpenFactory cluster
root_path = os.environ.get("OPENFACTORY_ROOT_PATH")
if (root_path and self.app.config.get("APPLICATION_ROOT") == "/"):
self.app.config["APPLICATION_ROOT"] = root_path
# expose OpenFactory app inside Flask
self.app.ofa_app = self
# internal state
self._server = None
self._thread_exception = None
self.configure_routes()
[docs]
def create_flask_app(self) -> Flask:
"""
Create and configure the Flask application instance.
This method is called during initialization to create the
:class:`flask.Flask` application attached to the OpenFactory runtime.
Subclasses may override this method to customize the Flask application
before routes are registered via :meth:`configure_routes`.
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.
.. admonition:: Example
.. code-block:: python
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.
"""
return Flask(__class__.__module__)
def _thread_wrapper(self, target):
"""
Execute a function inside a managed background thread.
Any unhandled exception raised by the thread is captured and stored
so it can be propagated back into the main asyncio runtime.
Args:
target: Callable executed inside the thread.
"""
try:
target()
except Exception as e:
self.logger.exception("Flask thread crashed")
self._thread_exception = e
def _run_flask(self):
"""
Run the embedded Flask application using Werkzeug.
This starts a blocking WSGI server inside a dedicated background thread.
The server listens on all network interfaces (``0.0.0.0``) and uses the
port defined by the ``PORT`` environment variable, or ``4000`` if the
variable is not set.
Note:
- Logging of incoming HTTP requests is controlled by the
``log_http_requests`` constructor parameter.
- This method is intended for internal use and is called automatically
by :meth:`async_run`.
"""
port = int(os.getenv("PORT", "4000"))
handler = None if self.log_http_requests else QuietWSGIRequestHandler
self._server = make_server(
host="0.0.0.0",
port=port,
app=self.app,
request_handler=handler,
)
self.logger.info(f"Starting Flask server on port {port}")
self._server.serve_forever()
def _user_defined_async_main(self) -> bool:
"""
Check whether the subclass defines :meth:`async_main_loop`.
Returns:
bool: ``True`` if the subclass overrides :meth:`async_main_loop`,
otherwise ``False``.
"""
return type(self).async_main_loop is not OpenFactoryApp.async_main_loop
def _user_defined_main(self) -> bool:
"""
Check whether the subclass defines :meth:`main_loop`.
Returns:
bool: ``True`` if the subclass overrides :meth:`OpenFactoryApp.main_loop <openfactory.apps.ofaapp.OpenFactoryApp.main_loop>`,
otherwise ``False``.
"""
return type(self).main_loop is not OpenFactoryApp.main_loop
async def _run_openfactory(self):
"""
Execute the OpenFactory application logic.
This method ensures that only asynchronous execution is used.
If :meth:`main_loop` is defined, a runtime error is raised.
Behavior:
- If :meth:`async_main_loop` is defined, it is awaited.
- Otherwise, the coroutine waits indefinitely.
Raises:
RuntimeError: If a synchronous :meth:`main_loop` is defined.
"""
if self._user_defined_main():
raise RuntimeError(
"OpenFactoryFlaskApp does NOT support 'main_loop'. "
"Use 'async_main_loop' instead."
)
if self._user_defined_async_main():
await self.async_main_loop()
else:
# keep app alive
while True:
await asyncio.sleep(3600)
[docs]
async def async_run(self):
"""
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 :meth:`OpenFactoryApp.app_event_loop_stopped() <openfactory.apps.ofaapp.OpenFactoryApp.app_event_loop_stopped>`.
"""
self.welcome_banner()
self.avail = "AVAILABLE"
flask_thread = threading.Thread(
target=lambda: self._thread_wrapper(self._run_flask),
name="FlaskThread",
daemon=False
)
flask_thread.start()
task_ofa = asyncio.create_task(
self._run_openfactory(),
name="OpenFactoryTask"
)
try:
while True:
# propagate flask thread exception
if self._thread_exception:
raise self._thread_exception
# propagate async task exception
if task_ofa.done():
exc = task_ofa.exception()
if exc:
raise exc
break
await asyncio.sleep(1)
except asyncio.CancelledError:
self.logger.info("Application cancelled")
raise
except KeyboardInterrupt:
self.logger.info("KeyboardInterrupt received")
except Exception:
self.logger.exception("Application crashed")
finally:
self.logger.info("Shutting down application")
# stop flask server
if self._server:
self._server.shutdown()
# cancel async task
if not task_ofa.done():
task_ofa.cancel()
await asyncio.gather(
task_ofa,
return_exceptions=True
)
# cleanup
try:
self.shutdown()
except Exception:
self.logger.exception("Cleanup failed")
self.logger.info("Application shutdown complete")
[docs]
def run(self):
"""
Start the application using a synchronous entry point.
This method runs :meth:`async_run` inside an asyncio event loop.
"""
asyncio.run(self.async_run())
[docs]
async def async_main_loop(self) -> None:
"""
Default asynchronous main loop.
Keeps the application alive when no custom loop is provided.
Subclasses can override this method to implement background logic.
.. admonition:: Example
.. code-block:: python
async def async_main_loop(self):
while True:
await asyncio.sleep(5)
self.logger.info("Background task running")
"""
while True:
await asyncio.sleep(3600)
