Enterprise Flask Migration Toolkit

openbb_platform/core/openbb_core/api/app_loader.py

@@ -7,6 +7,7 @@
 from openbb_core.app.router import RouterLoader
 from openbb_core.provider.utils.errors import EmptyDataError, UnauthorizedError
 from pydantic import ValidationError
+from starlette.routing import Mount
 
 
 class AppLoader:
@@ -15,8 +16,30 @@ class AppLoader:
     @staticmethod
     def add_routers(app: FastAPI, routers: list[APIRouter | None], prefix: str):
         """Add routers."""
+
+        def _join_paths(p1: str, p2: str) -> str:
+            if not p1:
+                return p2 or "/"
+            if not p2:
+                return p1 or "/"
+            joined = p1.rstrip("/") + "/" + p2.lstrip("/")
+            return joined.rstrip("/") or "/"
+
         for router in routers:
             if router:
+                # FastAPI's include_router doesn't propagate Starlette Mount routes.
+                # If an APIRouter contains mounted sub-apps (e.g. WSGIMiddleware for Flask),
+                # mount them directly on the FastAPI app with the same prefix.
+                for route in getattr(router, "routes", []):
+                    if not isinstance(route, Mount):
+                        continue
+                    mount_path = _join_paths(prefix, route.path)
+                    if any(
+                        isinstance(existing, Mount) and existing.path == mount_path
+                        for existing in app.router.routes
+                    ):
+                        continue
+                    app.mount(mount_path, route.app, name=route.name)
                 app.include_router(router=router, prefix=prefix)
 
     @staticmethod

openbb_platform/core/openbb_core/api/router/commands.py

@@ -23,6 +23,7 @@
 from openbb_core.env import Env
 from openbb_core.provider.utils.helpers import to_snake_case
 from pydantic import BaseModel
+from starlette.routing import Mount
 from typing_extensions import ParamSpec
 
 try:
@@ -347,7 +348,18 @@ def add_command_map(command_runner: CommandRunner, api_router: APIRouter) -> Non
     plugins_router = RouterLoader.from_extensions()
 
     for route in plugins_router.api_router.routes:
-        route.endpoint = build_api_wrapper(command_runner=command_runner, route=route)  # type: ignore # noqa
+        if isinstance(route, APIRoute):
+            route.endpoint = build_api_wrapper(command_runner=command_runner, route=route)  # type: ignore # noqa
+            continue
+        # Mounted sub-apps (e.g. WSGIMiddleware for Flask) are Starlette Mount routes.
+        # APIRouter.include_router will not carry these over, so we mount them manually.
+        if isinstance(route, Mount):
+            if any(
+                isinstance(existing, Mount) and existing.path == route.path
+                for existing in api_router.routes
+            ):
+                continue
+            api_router.mount(route.path, route.app, name=route.name)
     api_router.include_router(router=plugins_router.api_router)
 
 

openbb_platform/core/openbb_core/app/extension_loader.py

@@ -175,6 +175,23 @@ def load_core(eps: EntryPoints) -> dict[str, "Router"]:
                     entry = entry.router
                 if isinstance(entry, APIRouter):
                     entries[ep.name] = Router.from_fastapi(entry)
+                    continue
+                if "flask" in str(type(entry)).lower():
+                    try:
+                        import flask  # noqa: F401
+                    except ImportError:
+                        continue
+                    from openbb_core.app.utils.flask import FlaskExtensionLoader
+
+                    try:
+                        flask_extension = FlaskExtensionLoader.load_flask_extension(
+                            ep.value, ep.name
+                        )
+                        if flask_extension:
+                            entries[ep.name] = flask_extension
+                    except (ModuleNotFoundError, ImportError):
+                        continue
+
             return entries
 
         def load_provider(eps: EntryPoints) -> dict[str, "Provider"]:

openbb_platform/core/openbb_core/app/router.py

@@ -15,6 +15,7 @@
 )
 
 from fastapi import APIRouter, Depends
+from fastapi.routing import APIRoute
 from openbb_core.app.deprecation import DeprecationSummary, OpenBBDeprecationWarning
 from openbb_core.app.extension_loader import ExtensionLoader
 from openbb_core.app.model.abstract.warning import OpenBBWarning
@@ -28,6 +29,7 @@
 )
 from openbb_core.env import Env
 from pydantic import BaseModel
+from starlette.routing import Mount
 from typing_extensions import ParamSpec
 
 P = ParamSpec("P")
@@ -182,6 +184,29 @@ def include_router(
             prefix=prefix,
             tags=tags,  # type: ignore
         )
+
+        # FastAPI's APIRouter.include_router only includes APIRoute instances.
+        # Starlette Mount routes (used by .mount, e.g. for WSGIMiddleware) must
+        # be manually propagated, otherwise mounted apps silently disappear.
+        def _join_paths(p1: str, p2: str) -> str:
+            if not p1:
+                return p2 or "/"
+            if not p2:
+                return p1 or "/"
+            joined = p1.rstrip("/") + "/" + p2.lstrip("/")
+            return joined.rstrip("/") or "/"
+
+        for route in router.api_router.routes:
+            if not isinstance(route, Mount):
+                continue
+            mount_path = _join_paths(prefix, route.path)
+            if any(
+                isinstance(existing, Mount) and existing.path == mount_path
+                for existing in self._api_router.routes
+            ):
+                continue
+            self._api_router.mount(mount_path, route.app, name=route.name)
+
         name = prefix if prefix else router.prefix
         self._routers[name.strip("/")] = router
 
@@ -426,7 +451,11 @@ def get_command_map(
     ) -> dict[str, Callable]:
         """Get command map."""
         api_router = router.api_router
-        command_map = {route.path: route.endpoint for route in api_router.routes}  # type: ignore
+        command_map = {
+            route.path: route.endpoint
+            for route in api_router.routes
+            if isinstance(route, APIRoute)
+        }
         return command_map
 
     @staticmethod
@@ -440,6 +469,8 @@ def get_provider_coverage(
 
         coverage_map: dict[Any, Any] = {}
         for route in api_router.routes:
+            if not isinstance(route, APIRoute):
+                continue
             openapi_extra = getattr(route, "openapi_extra", None)
             if openapi_extra:
                 model = openapi_extra.get("model", None)
@@ -471,7 +502,9 @@ def get_command_coverage(
 
         coverage_map: dict[Any, Any] = {}
         for route in api_router.routes:
-            openapi_extra = getattr(route, "openapi_extra")
+            if not isinstance(route, APIRoute):
+                continue
+            openapi_extra = getattr(route, "openapi_extra", None)
             if openapi_extra:
                 model = openapi_extra.get("model", None)
                 if model:
@@ -493,7 +526,9 @@ def get_commands_model(router: Router, sep: str | None = None) -> dict[str, str]
 
         coverage_map: dict[Any, Any] = {}
         for route in api_router.routes:
-            openapi_extra = getattr(route, "openapi_extra")
+            if not isinstance(route, APIRoute):
+                continue
+            openapi_extra = getattr(route, "openapi_extra", None)
             if openapi_extra:
                 model = openapi_extra.get("model", None)
                 if model and hasattr(route, "path"):

openbb_platform/core/openbb_core/app/utils/flask/__init__.py

@@ -0,0 +1,39 @@
+"""Flask integration utilities for OpenBB Core.
+
+This module provides Flask app integration capabilities.
+All imports are lazy to avoid ImportError when Flask is not installed.
+"""
+
+
+def __getattr__(name: str):
+    """Lazy import to avoid ImportError when Flask is not installed."""
+    if name == "FlaskExtensionLoader":
+        from .loader import FlaskExtensionLoader
+
+        return FlaskExtensionLoader
+    if name == "FlaskIntrospector":
+        from .introspection import FlaskIntrospector
+
+        return FlaskIntrospector
+    if name == "OpenAPISpecGenerator":
+        from .adapter import OpenAPISpecGenerator
+
+        return OpenAPISpecGenerator
+    if name == "FlaskMountRegistry":
+        from .registry import FlaskMountRegistry
+
+        return FlaskMountRegistry
+    if name == "_check_flask_available":
+        from .introspection import _check_flask_available
+
+        return _check_flask_available
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+__all__ = [
+    "FlaskIntrospector",
+    "FlaskExtensionLoader",
+    "OpenAPISpecGenerator",
+    "FlaskMountRegistry",
+    "_check_flask_available",
+]

openbb_platform/core/openbb_core/app/utils/flask/adapter.py

@@ -0,0 +1,94 @@
+"""Flask-to-OpenBB conversion logic."""
+
+from typing import Any, Dict, List
+from .introspection import FlaskIntrospector, FlaskRouteInfo
+
+
+def is_flask_available() -> bool:
+    """Check if Flask is available."""
+    try:
+        import flask
+        return True
+    except ImportError:
+        return False
+
+
+class OpenAPISpecGenerator:
+    """Generate OpenAPI specification from Flask route metadata."""
+
+    TYPE_MAP = {
+        'str': 'string',
+        'int': 'integer',
+        'float': 'number',
+        'bool': 'boolean',
+    }
+
+    @classmethod
+    def generate_spec(cls, routes: List[FlaskRouteInfo]) -> Dict[str, Any]:
+        """Generate OpenAPI spec dictionary from Flask routes."""
+        paths: Dict[str, Any] = {}
+        schemas: Dict[str, Any] = {}
+
+        for route in routes:
+            openapi_path = cls._convert_flask_path(route['path'])
+            if openapi_path not in paths:
+                paths[openapi_path] = {}
+
+            for method in route['methods']:
+                operation = cls._generate_operation(route, schemas)
+                paths[openapi_path][method.lower()] = operation
+
+        return {
+            'paths': paths,
+            'components': {'schemas': schemas}
+        }
+
+    @classmethod
+    def _convert_flask_path(cls, flask_path: str) -> str:
+        """Convert Flask path format to OpenAPI format."""
+        import re
+        return re.sub(r'<(?:int:)?(?:float:)?(?:path:)?([^>]+)>', r'{\1}', flask_path)
+
+    @classmethod
+    def _generate_operation(cls, route: FlaskRouteInfo, schemas: Dict[str, Any]) -> Dict[str, Any]:
+        """Generate OpenAPI operation object for a route."""
+        operation: Dict[str, Any] = {
+            'summary': route['summary'] or f"{route['function_name']}",
+            'operationId': f"{route['function_name']}_{route['path'].replace('/', '_').strip('_')}",
+            'parameters': [],
+            'responses': {
+                '200': {'description': 'Successful Response'},
+                '400': {'description': 'Bad Request'},
+            }
+        }
+
+        if route['description']:
+            operation['description'] = route['description']
+
+        for param in route['path_params']:
+            operation['parameters'].append(cls._generate_parameter(param, 'path'))
+
+        for param in route['query_params']:
+            operation['parameters'].append(cls._generate_parameter(param, 'query'))
+
+        return operation
+
+    @classmethod
+    def _generate_parameter(cls, param: Dict[str, Any], location: str) -> Dict[str, Any]:
+        """Generate OpenAPI parameter object."""
+        param_spec: Dict[str, Any] = {
+            'name': param['name'],
+            'in': location,
+            'required': param['required'],
+            'schema': {
+                'type': cls.TYPE_MAP.get(param['type'], 'string'),
+            }
+        }
+
+        if param.get('description'):
+            param_spec['description'] = param['description']
+
+        if not param['required'] and param.get('default') is not None:
+            param_spec['schema']['default'] = param['default']
+
+        return param_spec