Decrease initial load on python package

flowfile/flowfile/__init__.py

@@ -54,8 +54,6 @@
     Utf8,
 )
 
-from flowfile.api import open_graph_in_editor
-from flowfile.web import start_server as start_web_ui
 from flowfile_core.flowfile import node_designer
 from flowfile_core.flowfile.flow_data_engine.flow_data_engine import FlowDataEngine
 from flowfile_core.flowfile.flow_data_engine.flow_file_column.main import FlowfileColumn
@@ -252,4 +250,21 @@
     "Field",
     "start_web_ui",
 ]
+
+
+def __getattr__(name: str):
+    # The web UI / editor entry points pull the FastAPI server stack (uvicorn,
+    # fastapi) and `requests`; resolve them lazily so a plain `import flowfile`
+    # for the dataframe API stays light.
+    if name == "open_graph_in_editor":
+        from flowfile.api import open_graph_in_editor
+
+        return open_graph_in_editor
+    if name == "start_web_ui":
+        from flowfile.web import start_server as start_web_ui
+
+        return start_web_ui
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
 logging.getLogger("PipelineHandler").setLevel(logging.WARNING)

flowfile/flowfile/__main__.py

@@ -20,6 +20,12 @@ def run_flow(flow_path: str, param_overrides: list[str] | None = None, run_id: i
 
     OFFLOAD_TO_WORKER.set(False)
 
+    # DB init is no longer an import side effect; get_run_user_id below reads the
+    # catalog DB via shared models (bypassing core's get_db guard), so init here.
+    from flowfile_core.database.connection import ensure_db_initialized
+
+    ensure_db_initialized()
+
     from flowfile_core.flowfile.manage.io_flowfile import open_flow
 
     path = Path(flow_path)

flowfile_core/flowfile_core/__init__.py

@@ -6,14 +6,11 @@
 from flowfile_core.utils.validate_setup import validate_setup
 
 validate_setup()
-from flowfile_core.database.init_db import init_db
 from flowfile_core.flowfile.handler import FlowfileHandler
 
 if "FLOWFILE_MODE" not in os.environ:
     os.environ["FLOWFILE_MODE"] = "electron"
 
-init_db()
-
 
 class ServerRun:
     exit: bool = False

flowfile_core/flowfile_core/ai/audit.py

@@ -28,7 +28,7 @@
 
 from sqlalchemy.orm import Session
 
-from flowfile_core.database.connection import SessionLocal
+from flowfile_core.database.connection import SessionLocal, ensure_db_initialized
 from flowfile_core.database.models import AiAuditEvent
 
 logger = logging.getLogger(__name__)
@@ -101,6 +101,7 @@ def record_event(event: AuditEvent, db: Session | None = None) -> AiAuditEvent:
         db.add(row)
         db.flush()
     else:
+        ensure_db_initialized()
         with SessionLocal() as session:
             session.add(row)
             session.commit()
@@ -140,6 +141,7 @@ def _apply(session: Session) -> AiAuditEvent | None:
 
     if db is not None:
         return _apply(db)
+    ensure_db_initialized()
     with SessionLocal() as session:
         row = _apply(session)
         if row is None:
@@ -184,6 +186,7 @@ def _query(session: Session) -> list[AiAuditEvent]:
 
     if db is not None:
         return _query(db)
+    ensure_db_initialized()
     with SessionLocal() as session:
         return _query(session)
 

flowfile_core/flowfile_core/ai/diff.py

@@ -610,8 +610,7 @@ def apply_diff(flow, diff: GraphDiff) -> ApplyResult:
         # AI modules (executor, audit, providers) deliberately keep out of
         # their import-time graph; this ``diff`` module imports lightly so
         # tests can stub ``flow_graph`` symbols where needed.
-        from fastapi import HTTPException
-
+        from flowfile_core.exceptions import FlowfileHTTPException
         from flowfile_core.flowfile.flow_graph import add_connection, delete_connection
 
         for add in diff.additions:
@@ -670,7 +669,7 @@ def apply_diff(flow, diff: GraphDiff) -> ApplyResult:
             connection = input_schema.NodeConnection.model_validate(c.connection)
             try:
                 delete_connection(flow, connection)
-            except HTTPException as exc:
+            except FlowfileHTTPException as exc:
                 # Narrow catch: only the "Connection does not exist" 422 gets
                 # swallowed. ``add_connection``'s cycle-detection 422 lives in
                 # the other loop and is unaffected; any other HTTPException

flowfile_core/flowfile_core/ai/diff_routes.py

@@ -41,7 +41,7 @@
 from flowfile_core import flow_file_handler
 from flowfile_core.ai import audit, diff
 from flowfile_core.auth.jwt import get_current_active_user
-from flowfile_core.database.connection import SessionLocal, get_db
+from flowfile_core.database.connection import SessionLocal, ensure_db_initialized, get_db
 
 router = APIRouter()
 
@@ -165,6 +165,7 @@ def _flip_audit_actions(audit_ids: list[int], action: audit.DiffAction) -> list[
     if not audit_ids:
         return []
     updated: list[int] = []
+    ensure_db_initialized()
     with SessionLocal() as session:
         for aid in audit_ids:
             row = audit.update_diff_action(aid, action, db=session)

flowfile_core/flowfile_core/ai/metrics.py

@@ -38,7 +38,7 @@
 from sqlalchemy.orm import Session
 
 from flowfile_core.ai.audit import query_events
-from flowfile_core.database.connection import SessionLocal
+from flowfile_core.database.connection import SessionLocal, ensure_db_initialized
 from flowfile_core.database.models import AiAuditEvent
 
 logger = logging.getLogger(__name__)
@@ -84,6 +84,7 @@ def _aggregate(session: Session) -> dict[str, float | int]:
 
     if db is not None:
         return _aggregate(db)
+    ensure_db_initialized()
     with SessionLocal() as session:
         return _aggregate(session)
 

flowfile_core/flowfile_core/ai/tools/executor/handlers/connections.py

@@ -422,13 +422,12 @@ def _handle_delete_connection(
             staged_node_payload={"delete_connection": connection.model_dump()},
         )
 
-    from fastapi import HTTPException
-
+    from flowfile_core.exceptions import FlowfileHTTPException
     from flowfile_core.flowfile.flow_graph import delete_connection
 
     try:
         delete_connection(flow, connection)
-    except HTTPException as exc:
+    except FlowfileHTTPException as exc:
         # LLM-redundant-op tolerance: swallow only the "Connection does not
         # exist" 422 (an ``update_node_settings`` in the same diff already
         # implicitly rewired, so this delete targets a wire that's already

flowfile_core/flowfile_core/auth/utils.py

@@ -7,8 +7,9 @@ def get_local_user_id() -> int:
     """Resolve the local_user's ID from the database for CLI execution."""
     try:
         from flowfile_core.database import models as db_models
-        from flowfile_core.database.connection import SessionLocal
+        from flowfile_core.database.connection import SessionLocal, ensure_db_initialized
 
+        ensure_db_initialized()
         db = SessionLocal()
         try:
             local_user = db.query(db_models.User).filter(db_models.User.username == "local_user").first()

flowfile_core/flowfile_core/catalog/delta_utils.py

@@ -12,7 +12,6 @@
 from pathlib import Path
 
 import pyarrow as pa
-from deltalake import DeltaTable
 
 from shared.delta_models import SourceTableVersion
 from shared.delta_utils import get_delta_size_bytes
@@ -40,6 +39,8 @@ def check_source_versions_current(source_table_versions_json: str | None) -> boo
         logger.warning("Could not parse source_table_versions JSON, treating as stale")
         return False
 
+    from deltalake import DeltaTable
+
     for sv in versions:
         try:
             current_version = DeltaTable(sv.file_path, without_files=True).version()
@@ -85,6 +86,8 @@ def get_delta_table_size_bytes(path: str | Path) -> int:
 
 def read_delta_preview(path: str, n_rows: int = 100) -> pa.Table:
     """Read the first N rows from a Delta table using PyArrow."""
+    from deltalake import DeltaTable
+
     dt = DeltaTable(str(path))
 
     dataset = dt.to_pyarrow_dataset()

flowfile_core/flowfile_core/catalog/migrate_parquet_to_delta.py

@@ -23,7 +23,7 @@
 import polars as pl
 
 from flowfile_core.catalog.delta_utils import is_delta_table
-from flowfile_core.database.connection import SessionLocal
+from flowfile_core.database.connection import SessionLocal, ensure_db_initialized
 from flowfile_core.database.models import CatalogTable
 
 logger = logging.getLogger(__name__)
@@ -88,6 +88,7 @@ def migrate_table(table: CatalogTable, *, dry_run: bool = False) -> bool:
 
 
 def main(dry_run: bool = False) -> int:
+    ensure_db_initialized()
     db = SessionLocal()
     try:
         tables = db.query(CatalogTable).filter(CatalogTable.storage_format == "parquet").all()

flowfile_core/flowfile_core/catalog/services/previews.py

@@ -6,7 +6,6 @@
 from pathlib import Path
 
 import polars as pl
-from deltalake import DeltaTable
 
 from flowfile_core.catalog.delta_utils import (
     is_delta_table,
@@ -170,6 +169,8 @@ def _get_delta_version_preview(self, data_path: Path, version: int, limit: int)
             except (RuntimeError, OSError, ValueError, KeyError):
                 logger.warning("Worker delta version preview failed, falling back to local", exc_info=True)
 
+        from deltalake import DeltaTable
+
         delta_table = DeltaTable(table_path, version=version)
         dataset = delta_table.to_pyarrow_dataset()
         pa_table = dataset.head(limit)
@@ -195,6 +196,8 @@ def get_table_history(self, table_id: int, limit: int | None = None) -> DeltaTab
             except (RuntimeError, OSError, ValueError, KeyError):
                 logger.warning("Worker delta history read failed, falling back to local", exc_info=True)
 
+        from deltalake import DeltaTable
+
         delta_table = DeltaTable(table_path, without_files=True)
         raw_history = delta_table.history(limit)
         current_version = delta_table.version()

flowfile_core/flowfile_core/catalog/services/tables.py

@@ -8,8 +8,6 @@
 from pathlib import Path
 from uuid import uuid4
 
-from deltalake import DeltaTable
-from pyarrow import dataset as ds
 from sqlalchemy.orm.attributes import flag_modified
 
 from flowfile_core.catalog.delta_utils import (
@@ -201,12 +199,16 @@ def _read_table_metadata(table_path: str, storage_format: str) -> tuple[list[dic
         path = Path(table_path)
 
         if storage_format == "delta" or (storage_format is None and is_delta_table(path)):
+            from deltalake import DeltaTable
+
             delta_table = DeltaTable(str(path))
             pa_schema = delta_table.schema().to_arrow()
             schema_list = [{"name": field.name, "dtype": str(field.type)} for field in pa_schema]
             row_count = delta_table.to_pyarrow_dataset().count_rows()
             size_bytes = get_delta_table_size_bytes(path)
         else:
+            from pyarrow import dataset as ds
+
             dataset = ds.dataset(str(path), format="parquet")
             schema_list = [{"name": field.name, "dtype": str(field.type)} for field in dataset.schema]
             row_count = dataset.count_rows()

flowfile_core/flowfile_core/database/connection.py

@@ -1,4 +1,6 @@
+import os
 import sys
+import threading
 from contextlib import contextmanager
 from pathlib import Path
 
@@ -27,9 +29,38 @@ def get_database_path() -> Path | None:
 
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
+_DB_INIT_LOCK = threading.Lock()
+_db_initialized = False
+
+
+def ensure_db_initialized() -> None:
+    """Idempotently run the startup Alembic migration + seed default rows.
+
+    Previously this ran as a side effect of ``import flowfile_core`` (via
+    ``init_db.py``), which made importing the dataframe API drag in Alembic and
+    create the catalog DB on disk. It is now deferred to first actual DB access
+    (``get_db``/``get_db_context``) and to explicit server startup. Cheap on
+    every call after the first. Honors ``FLOWFILE_SKIP_STARTUP_MIGRATION``.
+    """
+    global _db_initialized
+    if _db_initialized:
+        return
+    with _DB_INIT_LOCK:
+        if _db_initialized:
+            return
+        if not os.environ.get("FLOWFILE_SKIP_STARTUP_MIGRATION"):
+            from flowfile_core.database.migration import run_startup_migration
+
+            run_startup_migration()
+        from flowfile_core.database.init_db import init_db
+
+        init_db()
+        _db_initialized = True
+
 
 def get_db():
     """Dependency for FastAPI to get database session."""
+    ensure_db_initialized()
     db = SessionLocal()
     try:
         yield db
@@ -40,6 +71,7 @@ def get_db():
 @contextmanager
 def get_db_context():
     """Context manager for getting database session."""
+    ensure_db_initialized()
     db = SessionLocal()
     try:
         yield db

flowfile_core/flowfile_core/database/init_db.py

@@ -10,7 +10,6 @@
 from flowfile_core.auth.password import get_password_hash
 from flowfile_core.database import models as db_models
 from flowfile_core.database.connection import SessionLocal
-from flowfile_core.database.migration import run_startup_migration
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
@@ -20,12 +19,6 @@
 
 logger = logging.getLogger(__name__)
 
-# Run Alembic-based migrations (replaces the old manual run_migrations + create_all).
-# Skipped when FLOWFILE_SKIP_STARTUP_MIGRATION is set so the alembic CLI can import
-# our metadata without recursively re-entering migration machinery.
-if not os.environ.get("FLOWFILE_SKIP_STARTUP_MIGRATION"):
-    run_startup_migration()
-
 
 def create_default_local_user(db: Session):
     local_user = db.query(db_models.User).filter(db_models.User.username == "local_user").first()
@@ -184,5 +177,7 @@ def init_db():
 
 
 if __name__ == "__main__":
-    init_db()
+    from flowfile_core.database.connection import ensure_db_initialized
+
+    ensure_db_initialized()
     print("Local user created successfully")

flowfile_core/flowfile_core/exceptions.py

@@ -0,0 +1,12 @@
+class FlowfileHTTPException(Exception):
+    """HTTP-style error raised by engine/secret code without importing FastAPI.
+
+    A handler registered in ``flowfile_core.main`` maps this to the same JSON
+    response FastAPI's built-in ``HTTPException`` produces, so route behaviour is
+    unchanged while keeping ``flowfile_core.flowfile`` importable without FastAPI.
+    """
+
+    def __init__(self, status_code: int, detail: str | None = None):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(detail)

flowfile_core/flowfile_core/flowfile/flow_data_engine/sample_data.py

@@ -5,7 +5,6 @@
 from typing import Any
 
 import polars as pl
-from faker import Faker
 
 
 def create_fake_data(n_records: int = 1000, optimized: bool = True) -> pl.DataFrame:
@@ -18,6 +17,8 @@ def create_fake_data(n_records: int = 1000, optimized: bool = True) -> pl.DataFr
     Returns:
         pl.DataFrame
     """
+    from faker import Faker
+
     fake = Faker()
     selector = partial(randint, 0)
 
@@ -78,6 +79,8 @@ def generate_phone_number():
 def create_fake_data_raw(
     n_records: int = 1000, col_selection: list[str] = None
 ) -> Generator[dict[str, Any], None, None]:
+    from faker import Faker
+
     fake = Faker()
     selector = partial(randint, 0)