OpenFactoryFastAPIApp#

class openfactory.apps.ofa_fastapi_app.OpenFactoryFastAPIApp(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 FastAPI web interface.

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

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

Runtime behavior:

Calling run() starts both the OpenFactory application and an embedded HTTP server.
The HTTP server is powered by Uvicorn and serves the attached fastapi.FastAPI application.
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.
The server log level is aligned with the OpenFactory application logger.
HTTP access logging can be enabled or disabled using the log_http_requests constructor parameter.

api#

FastAPI application instance attached to the OpenFactory app.

Type:: fastapi.FastAPI

Usage Example (inline routes)

import os
from openfactory.apps import OpenFactoryFastAPIApp, EventAttribute, ofa_method
from openfactory.kafka import KSQLDBClient

class DemoFastAPIApp(OpenFactoryFastAPIApp):

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

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        @self.api.get("/")
        async def root():
            return {"status": self.status.value}

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

app = DemoFastAPIApp(
    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 included using routers:

Using routers (recommended for larger applications)

# main.py
import os
from openfactory.apps import OpenFactoryFastAPIApp, EventAttribute, ofa_method
from openfactory.kafka import KSQLDBClient
from routes import root, move

class DemoFastAPIApp(OpenFactoryFastAPIApp):

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

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # expose OpenFactory App inside FastAPI
        self.api.state.ofa_app = self

        # include routers
        self.api.include_router(root.router)
        self.api.include_router(move.router)

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

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

# routes/root.py
from fastapi import APIRouter, Request

router = APIRouter()

@router.get("/")
async def root(request: Request):
    ofa_app = request.app.state.ofa_app
    return {"status": ofa_app.status.value}

# routes/move.py
from fastapi import APIRouter, Request

router = APIRouter()

@router.post("/move")
async def move(x: float, y: float, request: Request):
    ofa_app = request.app.state.ofa_app
    ofa_app.move_axis(x, y)
    return {"message": "moving"}

Prometheus metrics can be exposed by calling expose_metrics(). By default, the global Prometheus registry is exposed, allowing both OpenFactory metrics and application-defined metrics to be scraped through the same endpoint.

Exposing Prometheus metrics

import asyncio
from prometheus_client import Counter
from openfactory.apps import OpenFactoryFastAPIApp

REQUESTS = Counter(
    "requests_total",
    "Number of processed requests"
)

class DemoFastAPIApp(OpenFactoryFastAPIApp):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.expose_metrics()

    async def async_main_loop(self):
        while True:
            REQUESTS.inc()
            await asyncio.sleep(1)

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

Note

The FastAPI application is accessible via api and behaves like a standard FastAPI instance.
Route definitions can be added either directly using api or via api.include_router().
For small applications, routes can be defined inline; for larger applications, using routers is recommended.
Only asynchronous execution is supported. Subclasses may optionally 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.

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 FastAPI server and the OpenFactory logic concurrently.

Return type:: None

This method:

Displays the welcome banner
Sets the application availability
Runs FastAPI and OpenFactory tasks concurrently

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

configure_routes()[source]#

Configure HTTP routes for the FastAPI application.

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

Example

def configure_routes(self):

    @self.api.get("/")
    async def root():
        return {"status": "ok"}

Note

For larger applications, prefer using api.include_router() instead of overriding this method.

Return type:: None

expose_metrics(path='/metrics', registry=<prometheus_client.registry.CollectorRegistry object>)[source]#

Expose Prometheus metrics through the embedded FastAPI application.

Registers an HTTP endpoint that returns all metrics contained in the specified Prometheus registry using the standard Prometheus exposition format.

Calling this method also registers the metrics endpoint with the OpenFactory platform, allowing other OpenFactory components to discover and scrape it automatically.

By default, the global Prometheus registry is exposed, allowing both OpenFactory metrics and user-defined Prometheus metrics to be scraped through the same endpoint.

This method is idempotent. Calling it multiple times with the same path has no effect.

Return type:

None

Parameters:

path – URL path where metrics are exposed. Defaults to "/metrics".
registry – Prometheus registry to expose. Defaults to the global Prometheus registry.

Raises:

ValueError – If another endpoint is already registered for path.

run()[source]#

Start the application using a synchronous entry point.

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

Return type:: None