diff --git a/flowfile/flowfile/__init__.py b/flowfile/flowfile/__init__.py
index b76b4b2ea..674c2a1a8 100644
--- a/flowfile/flowfile/__init__.py
+++ b/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)
diff --git a/flowfile/flowfile/__main__.py b/flowfile/flowfile/__main__.py
index 2f613c6d1..be14613b6 100644
--- a/flowfile/flowfile/__main__.py
+++ b/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)
diff --git a/flowfile_core/flowfile_core/__init__.py b/flowfile_core/flowfile_core/__init__.py
index a2a7e0767..73b528cbd 100644
--- a/flowfile_core/flowfile_core/__init__.py
+++ b/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
diff --git a/flowfile_core/flowfile_core/ai/audit.py b/flowfile_core/flowfile_core/ai/audit.py
index b16a6ec7f..5a05c7fd1 100644
--- a/flowfile_core/flowfile_core/ai/audit.py
+++ b/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)
 
diff --git a/flowfile_core/flowfile_core/ai/diff.py b/flowfile_core/flowfile_core/ai/diff.py
index e9a7c2b60..60274eec4 100644
--- a/flowfile_core/flowfile_core/ai/diff.py
+++ b/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
diff --git a/flowfile_core/flowfile_core/ai/diff_routes.py b/flowfile_core/flowfile_core/ai/diff_routes.py
index 015d7fdc8..a051bbfb4 100644
--- a/flowfile_core/flowfile_core/ai/diff_routes.py
+++ b/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)
diff --git a/flowfile_core/flowfile_core/ai/metrics.py b/flowfile_core/flowfile_core/ai/metrics.py
index 033a4b58e..c6d573094 100644
--- a/flowfile_core/flowfile_core/ai/metrics.py
+++ b/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)
 
diff --git a/flowfile_core/flowfile_core/ai/tools/executor/handlers/connections.py b/flowfile_core/flowfile_core/ai/tools/executor/handlers/connections.py
index fc08e0e13..c30425e53 100644
--- a/flowfile_core/flowfile_core/ai/tools/executor/handlers/connections.py
+++ b/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
diff --git a/flowfile_core/flowfile_core/auth/utils.py b/flowfile_core/flowfile_core/auth/utils.py
index 5ea371060..d2e59c545 100644
--- a/flowfile_core/flowfile_core/auth/utils.py
+++ b/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()
diff --git a/flowfile_core/flowfile_core/catalog/delta_utils.py b/flowfile_core/flowfile_core/catalog/delta_utils.py
index 6f426b9cd..90665f588 100644
--- a/flowfile_core/flowfile_core/catalog/delta_utils.py
+++ b/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()
diff --git a/flowfile_core/flowfile_core/catalog/migrate_parquet_to_delta.py b/flowfile_core/flowfile_core/catalog/migrate_parquet_to_delta.py
index f08e37a39..50445ea9c 100644
--- a/flowfile_core/flowfile_core/catalog/migrate_parquet_to_delta.py
+++ b/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()
diff --git a/flowfile_core/flowfile_core/catalog/services/previews.py b/flowfile_core/flowfile_core/catalog/services/previews.py
index e6c040ae2..841d55aab 100644
--- a/flowfile_core/flowfile_core/catalog/services/previews.py
+++ b/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()
diff --git a/flowfile_core/flowfile_core/catalog/services/tables.py b/flowfile_core/flowfile_core/catalog/services/tables.py
index 7fde04a18..1efb38c84 100644
--- a/flowfile_core/flowfile_core/catalog/services/tables.py
+++ b/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()
diff --git a/flowfile_core/flowfile_core/database/connection.py b/flowfile_core/flowfile_core/database/connection.py
index bfe6f31ba..597242611 100644
--- a/flowfile_core/flowfile_core/database/connection.py
+++ b/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
diff --git a/flowfile_core/flowfile_core/database/init_db.py b/flowfile_core/flowfile_core/database/init_db.py
index ac1ada206..104e61b4e 100644
--- a/flowfile_core/flowfile_core/database/init_db.py
+++ b/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")
diff --git a/flowfile_core/flowfile_core/exceptions.py b/flowfile_core/flowfile_core/exceptions.py
new file mode 100644
index 000000000..49f7681a6
--- /dev/null
+++ b/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)
diff --git a/flowfile_core/flowfile_core/flowfile/flow_data_engine/sample_data.py b/flowfile_core/flowfile_core/flowfile/flow_data_engine/sample_data.py
index 4c92f2584..6b497d071 100644
--- a/flowfile_core/flowfile_core/flowfile/flow_data_engine/sample_data.py
+++ b/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)
 
diff --git a/flowfile_core/flowfile_core/flowfile/flow_graph.py b/flowfile_core/flowfile_core/flowfile/flow_graph.py
index d56d0f0ce..455fec016 100644
--- a/flowfile_core/flowfile_core/flowfile/flow_graph.py
+++ b/flowfile_core/flowfile_core/flowfile/flow_graph.py
@@ -17,7 +17,6 @@
 import fastexcel
 import polars as pl
 import yaml
-from fastapi.exceptions import HTTPException
 from pyarrow.parquet import ParquetFile
 
 from flowfile_core.auth import sharing
@@ -30,6 +29,7 @@
 from flowfile_core.configs.node_store import CUSTOM_NODE_STORE
 from flowfile_core.database import models as db_models
 from flowfile_core.database.connection import get_db_context
+from flowfile_core.exceptions import FlowfileHTTPException
 from flowfile_core.flowfile.analytics.utils import create_graphic_walker_node_from_node_promise
 from flowfile_core.flowfile.artifacts import ArtifactContext
 from flowfile_core.flowfile.database_connection_manager.db_connections import (
@@ -348,7 +348,7 @@ def get_cloud_connection_settings(
         A FullCloudStorageConnection object with the connection details.
 
     Raises:
-        HTTPException: If the connection settings cannot be found.
+        FlowfileHTTPException: If the connection settings cannot be found.
     """
     cloud_connection_settings = get_local_cloud_connection(connection_name, user_id)
     if cloud_connection_settings is None and auth_mode in ("env_vars", transform_schema.AUTO_DATA_TYPE):
@@ -357,7 +357,7 @@ def get_cloud_connection_settings(
     elif cloud_connection_settings is None and auth_mode == "aws-cli":
         cloud_connection_settings = FullCloudStorageConnection(storage_type="s3", auth_method="aws-cli")
     if cloud_connection_settings is None:
-        raise HTTPException(status_code=400, detail="Cloud connection settings not found")
+        raise FlowfileHTTPException(status_code=400, detail="Cloud connection settings not found")
     return cloud_connection_settings
 
 
@@ -894,14 +894,14 @@ def _resolve_database_credentials(
         database_connection = database_settings.database_connection
         encrypted_password = get_encrypted_secret(current_user_id=user_id, secret_name=database_connection.password_ref)
         if encrypted_password is None:
-            raise HTTPException(status_code=400, detail="Password not found")
+            raise FlowfileHTTPException(status_code=400, detail="Password not found")
         return database_connection, encrypted_password, None
     elif is_sqlite:
         return database_settings.database_connection, None, None
     else:
         ref_settings = get_local_database_connection(database_settings.database_connection_name, user_id)
         if ref_settings is None:
-            raise HTTPException(
+            raise FlowfileHTTPException(
                 status_code=400,
                 detail=(
                     f"Database connection '{database_settings.database_connection_name}' not found "
@@ -3845,7 +3845,7 @@ def _get_kafka_read_settings() -> KafkaReadSettings:
                                     db, kafka_settings.kafka_connection_name, node_kafka_source.user_id
                                 )
                             if db_conn is None:
-                                raise HTTPException(status_code=400, detail="Kafka connection not found")
+                                raise FlowfileHTTPException(status_code=400, detail="Kafka connection not found")
                         consumer_config = build_consumer_config(db, db_conn, node_kafka_source.user_id)
                     _read_settings["v"] = KafkaReadSettings.from_consumer_config(
                         consumer_config,
@@ -3986,7 +3986,7 @@ def _build_worker_settings() -> WorkerGoogleAnalyticsReadSettings:
             with get_db_context() as db:
                 db_conn = get_ga_connection(db, ga_settings.ga_connection_name, node_ga_reader.user_id)
                 if db_conn is None:
-                    raise HTTPException(
+                    raise FlowfileHTTPException(
                         status_code=400,
                         detail=(
                             f"Google Analytics connection '{ga_settings.ga_connection_name}' not found "
@@ -3998,7 +3998,7 @@ def _build_worker_settings() -> WorkerGoogleAnalyticsReadSettings:
                     db, ga_settings.ga_connection_name, node_ga_reader.user_id
                 )
                 if encrypted_credential is None:
-                    raise HTTPException(
+                    raise FlowfileHTTPException(
                         status_code=400,
                         detail=(
                             f"Google Analytics connection '{ga_settings.ga_connection_name}' has no stored credential"
@@ -4042,7 +4042,7 @@ def _build_worker_settings() -> WorkerGoogleAnalyticsReadSettings:
             # ``oauth_cfg`` is only fetched for the oauth auth method; guard against
             # an unknown auth_method value reaching this branch with ``None``.
             if not oauth_cfg or not oauth_cfg["client_id"] or not oauth_cfg["client_secret"]:
-                raise HTTPException(
+                raise FlowfileHTTPException(
                     status_code=500,
                     detail=(
                         "Google OAuth is not configured on this instance. Open Admin → Google OAuth "
@@ -5476,9 +5476,9 @@ def add_connection(flow: FlowGraph, node_connection: input_schema.NodeConnection
             )
             if not n
         ]
-        raise HTTPException(404, f"Node(s) not found: {', '.join(missing)}")
+        raise FlowfileHTTPException(404, f"Node(s) not found: {', '.join(missing)}")
     if _would_create_cycle(from_node, to_node):
-        raise HTTPException(
+        raise FlowfileHTTPException(
             422,
             f"Connecting node {from_node.node_id} -> {to_node.node_id} would create a cycle",
         )
@@ -5502,13 +5502,13 @@ def delete_connection(graph, node_connection: input_schema.NodeConnection):
     # already removed) surfaces as an AttributeError → 500, which also drops
     # CORS headers and shows up as a CORS error in the browser.
     if from_node is None or to_node is None:
-        raise HTTPException(422, "Connection does not exist on the input node")
+        raise FlowfileHTTPException(422, "Connection does not exist on the input node")
     connection_valid = to_node.node_inputs.validate_if_input_connection_exists(
         node_input_id=from_node.node_id,
         connection_name=node_connection.input_connection.get_node_input_connection_type(),
     )
     if not connection_valid:
-        raise HTTPException(422, "Connection does not exist on the input node")
+        raise FlowfileHTTPException(422, "Connection does not exist on the input node")
     from_node.delete_lead_to_node(node_connection.input_connection.node_id)
     to_node.delete_input_node(
         node_connection.output_connection.node_id,
diff --git a/flowfile_core/flowfile_core/kernel/__init__.py b/flowfile_core/flowfile_core/kernel/__init__.py
index dcd57f0c8..f453550e4 100644
--- a/flowfile_core/flowfile_core/kernel/__init__.py
+++ b/flowfile_core/flowfile_core/kernel/__init__.py
@@ -16,7 +16,6 @@
     RecoveryMode,
     RecoveryStatus,
 )
-from flowfile_core.kernel.routes import router
 
 __all__ = [
     "KernelManager",
@@ -54,3 +53,15 @@ def get_kernel_manager() -> KernelManager:
         shared_path = str(storage.temp_directory / "kernel_shared")
         _manager = KernelManager(shared_volume_path=shared_path)
     return _manager
+
+
+def __getattr__(name: str):
+    # `router` is imported only by the FastAPI server (main.py). Loading it
+    # eagerly would pull fastapi into every importer of this package (incl.
+    # flowfile_frame via flow_graph's KernelManager/execution imports), so
+    # resolve it lazily to keep the dataframe-API import light.
+    if name == "router":
+        from flowfile_core.kernel.routes import router
+
+        return router
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
diff --git a/flowfile_core/flowfile_core/main.py b/flowfile_core/flowfile_core/main.py
index 09693921c..d27d5c114 100644
--- a/flowfile_core/flowfile_core/main.py
+++ b/flowfile_core/flowfile_core/main.py
@@ -9,8 +9,9 @@
 from pathlib import Path
 
 import uvicorn
-from fastapi import BackgroundTasks, FastAPI
+from fastapi import BackgroundTasks, FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
 
 from flowfile_core.ai import router as ai_router
 from flowfile_core.ai.admin_routes import router as ai_admin_router
@@ -23,6 +24,8 @@
     WORKER_PORT,
     WORKER_URL,
 )
+from flowfile_core.database.connection import ensure_db_initialized
+from flowfile_core.exceptions import FlowfileHTTPException
 from flowfile_core.kernel import router as kernel_router
 from flowfile_core.ml import router as ml_router
 from flowfile_core.routes.api_consumers import router as api_consumers_router
@@ -66,6 +69,10 @@ async def shutdown_handler(app: FastAPI):
 
     print("Starting core application...")
 
+    # DB init/migration used to run as an import side effect of flowfile_core;
+    # it's now deferred, so run it explicitly at server startup (idempotent).
+    ensure_db_initialized()
+
     # Only auto-start scheduler if explicitly opted in via env var
     if os.environ.get("FLOWFILE_SCHEDULER_ENABLED", "").lower() in ("true", "1", "yes"):
         scheduler = FlowScheduler()
@@ -120,6 +127,13 @@ def _shutdown_local_model():
     lifespan=shutdown_handler,
 )
 
+
+@app.exception_handler(FlowfileHTTPException)
+async def _flowfile_http_exception_handler(request: Request, exc: FlowfileHTTPException):
+    """Map engine/secret FlowfileHTTPExceptions to the same JSON response FastAPI's
+    built-in HTTPException handler produces (keeps flow_graph fastapi-import-free)."""
+    return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
+
 # The Tauri 2 desktop shell loads the renderer from a custom protocol — the
 # exact origin differs per OS (`tauri://localhost` on macOS/iOS, `http://tauri.localhost`
 # on Linux, `https://tauri.localhost` on Windows/Android). A regex covers all
@@ -271,6 +285,10 @@ def _run_flow_cli(flow_path: str, run_id: int) -> int:
 
     OFFLOAD_TO_WORKER.set(False)
 
+    # Standalone CLI run: no FastAPI lifespan, and get_run_user_id below reads the
+    # catalog DB via shared models (bypassing core's get_db guard), so init here.
+    ensure_db_initialized()
+
     from flowfile_core.flowfile.manage.io_flowfile import open_flow
 
     path = Path(flow_path)
diff --git a/flowfile_core/flowfile_core/secret_manager/secret_manager.py b/flowfile_core/flowfile_core/secret_manager/secret_manager.py
index 71f4ed234..9db9ed2d2 100644
--- a/flowfile_core/flowfile_core/secret_manager/secret_manager.py
+++ b/flowfile_core/flowfile_core/secret_manager/secret_manager.py
@@ -3,7 +3,6 @@
 from cryptography.fernet import Fernet
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.kdf.hkdf import HKDF
-from fastapi.exceptions import HTTPException
 from pydantic import SecretStr
 from sqlalchemy import and_
 from sqlalchemy.orm import Session
@@ -13,6 +12,7 @@
 from flowfile_core.auth.secrets import get_master_key
 from flowfile_core.database import models as db_models
 from flowfile_core.database.connection import get_db_context
+from flowfile_core.exceptions import FlowfileHTTPException
 
 # Version identifier for key derivation scheme (allows future migrations)
 KEY_DERIVATION_VERSION = b"flowfile-secrets-v1"
@@ -175,7 +175,7 @@ def delete_secret(db: Session, secret_name: str, user_id: int) -> None:
     )
 
     if not db_secret:
-        raise HTTPException(status_code=404, detail="Secret not found")
+        raise FlowfileHTTPException(status_code=404, detail="Secret not found")
 
     sharing.delete_grants_for_resource(db, "secret", db_secret.id)
     db.delete(db_secret)
diff --git a/flowfile_core/tests/conftest.py b/flowfile_core/tests/conftest.py
index f047d9855..c04c89813 100644
--- a/flowfile_core/tests/conftest.py
+++ b/flowfile_core/tests/conftest.py
@@ -61,11 +61,10 @@ def is_port_in_use(port, host='localhost'):
 @pytest.fixture(scope="session", autouse=True)
 def setup_test_db():
     """Setup the test database and clean up after tests"""
-    from flowfile_core.database.connection import engine, get_database_url
-    from flowfile_core.database.init_db import init_db
+    from flowfile_core.database.connection import engine, ensure_db_initialized, get_database_url
     from flowfile_core.database.models import Base
 
-    init_db()
+    ensure_db_initialized()
 
     yield
 
diff --git a/flowfile_core/tests/flowfile/test_flow_graph_utils.py b/flowfile_core/tests/flowfile/test_flow_graph_utils.py
index 35789e3b7..e44513bd0 100644
--- a/flowfile_core/tests/flowfile/test_flow_graph_utils.py
+++ b/flowfile_core/tests/flowfile/test_flow_graph_utils.py
@@ -1,7 +1,7 @@
 
 import pytest
 
-from fastapi import HTTPException
+from flowfile_core.exceptions import FlowfileHTTPException
 
 from flowfile_core.flowfile.flow_graph import FlowGraph, add_connection
 from flowfile_core.flowfile.flow_graph_utils import _create_node_id_mapping, _validate_input, combine_flow_graphs
@@ -456,7 +456,7 @@ def _build_chain_for_cycle_tests() -> FlowGraph:
 
 def test_add_connection_rejects_self_loop():
     graph = _build_chain_for_cycle_tests()
-    with pytest.raises(HTTPException) as exc:
+    with pytest.raises(FlowfileHTTPException) as exc:
         add_connection(graph, input_schema.NodeConnection.create_from_simple_input(2, 2))
     assert exc.value.status_code == 422
     assert "cycle" in exc.value.detail.lower()
@@ -465,7 +465,7 @@ def test_add_connection_rejects_self_loop():
 def test_add_connection_rejects_direct_back_edge():
     graph = _build_chain_for_cycle_tests()
     # 1 -> 2 exists; adding 2 -> 1 closes a 2-cycle.
-    with pytest.raises(HTTPException) as exc:
+    with pytest.raises(FlowfileHTTPException) as exc:
         add_connection(graph, input_schema.NodeConnection.create_from_simple_input(2, 1))
     assert exc.value.status_code == 422
 
@@ -473,7 +473,7 @@ def test_add_connection_rejects_direct_back_edge():
 def test_add_connection_rejects_transitive_back_edge():
     graph = _build_chain_for_cycle_tests()
     # Chain is 1 -> 2 -> 3; closing 3 -> 1 would form a 3-cycle.
-    with pytest.raises(HTTPException) as exc:
+    with pytest.raises(FlowfileHTTPException) as exc:
         add_connection(graph, input_schema.NodeConnection.create_from_simple_input(3, 1))
     assert exc.value.status_code == 422
 
diff --git a/flowfile_core/tests/test_lazy_imports.py b/flowfile_core/tests/test_lazy_imports.py
new file mode 100644
index 000000000..4766aae78
--- /dev/null
+++ b/flowfile_core/tests/test_lazy_imports.py
@@ -0,0 +1,76 @@
+"""Import-weight contract for the programmatic API.
+
+`import flowfile_frame` / `import flowfile` build ETL graphs in-process and must
+stay lightweight: they must NOT drag in the FastAPI server stack, Alembic, the
+cloud/data heavyweights (boto3, deltalake, gcsfs), or faker, and importing them
+must NOT create or migrate the catalog DB on disk (that is deferred to first
+actual DB access / server startup — see database/connection.ensure_db_initialized).
+
+Each check runs in a fresh subprocess because sys.modules is process-global: by
+the time the pytest session reaches this file, other test modules have already
+imported FastAPI, so an in-process assertion would be meaningless.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+# Server/heavy deps that must stay off the dataframe-API import path.
+BANNED = ["fastapi", "alembic", "faker", "boto3", "botocore", "deltalake", "gcsfs", "uvicorn"]
+
+
+def _run(script: str) -> str:
+    """Run a snippet in a fresh interpreter with a throwaway catalog DB path.
+
+    PYTHONPATH is seeded from the parent's ``sys.path`` so the subprocess imports
+    the exact same packages this test session does (the package-under-test), not
+    whatever a bare interpreter's site config happens to resolve.
+    """
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        env = {
+            **os.environ,
+            "FLOWFILE_DB_PATH": str(Path(tmp) / "catalog.db"),
+            "PYTHONPATH": os.pathsep.join(p for p in sys.path if p),
+        }
+        proc = subprocess.run(
+            [sys.executable, "-c", script],
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+        assert proc.returncode == 0, f"subprocess failed:\nstdout={proc.stdout}\nstderr={proc.stderr}"
+        return proc.stdout
+
+
+def test_import_flowfile_frame_is_lightweight() -> None:
+    out = _run(
+        "import os, sys\n"
+        f"banned = {BANNED!r}\n"
+        "import flowfile_frame\n"
+        "leaked = [m for m in banned if m in sys.modules]\n"
+        "assert not leaked, f'import flowfile_frame leaked heavy modules: {leaked}'\n"
+        "assert not os.path.exists(os.environ['FLOWFILE_DB_PATH']), 'import created the catalog DB on disk'\n"
+        "print('ok')\n"
+    )
+    assert "ok" in out
+
+
+def test_import_flowfile_is_lightweight() -> None:
+    out = _run(
+        "import os, sys\n"
+        f"banned = {BANNED!r}\n"
+        "import flowfile\n"
+        "leaked = [m for m in banned if m in sys.modules]\n"
+        "assert not leaked, f'import flowfile leaked heavy modules: {leaked}'\n"
+        "assert not os.path.exists(os.environ['FLOWFILE_DB_PATH']), 'import created the catalog DB on disk'\n"
+        # The web-UI entry points stay resolvable (lazily) without being loaded at import.
+        "assert callable(flowfile.start_web_ui)\n"
+        "assert callable(flowfile.open_graph_in_editor)\n"
+        "print('ok')\n"
+    )
+    assert "ok" in out
diff --git a/flowfile_frame/flowfile_frame/cloud_storage/secret_manager.py b/flowfile_frame/flowfile_frame/cloud_storage/secret_manager.py
index 0773174fb..1b6455707 100644
--- a/flowfile_frame/flowfile_frame/cloud_storage/secret_manager.py
+++ b/flowfile_frame/flowfile_frame/cloud_storage/secret_manager.py
@@ -1,4 +1,3 @@
-from flowfile_core.auth.jwt import create_access_token, get_current_user_sync
 from flowfile_core.database.connection import get_db_context
 from flowfile_core.flowfile.database_connection_manager.db_connections import (
     delete_cloud_connection,
@@ -9,6 +8,10 @@
 
 
 def get_current_user_id() -> int | None:
+    # Imported lazily: flowfile_core.auth.jwt is a FastAPI router module, so a
+    # top-level import would drag FastAPI into `import flowfile_frame`.
+    from flowfile_core.auth.jwt import create_access_token, get_current_user_sync
+
     access_token = create_access_token(data={"sub": "local_user"})
     with get_db_context() as db:
         current_user_id = get_current_user_sync(access_token, db).id
@@ -25,6 +28,8 @@ def create_cloud_storage_connection(connection: FullCloudStorageConnection) -> N
     Returns:
         None
     """
+    from flowfile_core.auth.jwt import create_access_token, get_current_user_sync
+
     access_token = create_access_token(data={"sub": "local_user"})
 
     with get_db_context() as db:
diff --git a/shared/cloud_storage/directory.py b/shared/cloud_storage/directory.py
index c4907708a..752990f40 100644
--- a/shared/cloud_storage/directory.py
+++ b/shared/cloud_storage/directory.py
@@ -8,9 +8,6 @@
 
 from typing import Any
 
-import boto3
-from botocore.exceptions import ClientError
-
 
 def get_first_file_from_cloud_dir(source: str, storage_options: dict[str, Any] | None = None) -> str:
     """Get the first file matching the extension from a cloud storage directory.
@@ -145,6 +142,8 @@ def _remove_wildcards_from_prefix(prefix: str) -> str:
 
 def _create_s3_client(storage_options: dict[str, Any] | None):
     """Create boto3 S3 client with optional credentials."""
+    import boto3
+
     if storage_options is None:
         return boto3.client("s3")
 
@@ -158,6 +157,8 @@ def _create_s3_client(storage_options: dict[str, Any] | None):
 
 def _get_first_file(s3_client, bucket_name: str, base_prefix: str, file_extension: str) -> dict[Any, Any]:
     """List objects and return the first file matching the extension."""
+    from botocore.exceptions import ClientError
+
     try:
         paginator = s3_client.get_paginator("list_objects_v2")
         pages = paginator.paginate(Bucket=bucket_name, Prefix=base_prefix)
diff --git a/shared/cloud_storage/gcs.py b/shared/cloud_storage/gcs.py
index 58f7d4c15..e318979bc 100644
--- a/shared/cloud_storage/gcs.py
+++ b/shared/cloud_storage/gcs.py
@@ -12,9 +12,7 @@
 from typing import Any, Literal
 from urllib.parse import urlparse
 
-import gcsfs
 import polars as pl
-from deltalake import DeltaTable, write_deltalake
 
 
 def use_pyarrow_for_gcs(storage_type: str, endpoint_url: str | None) -> bool:
@@ -80,6 +78,7 @@ def get_lazy_frame_from_gcs_pyarrow_dataset(
     is_directory
         If True, glob/wildcard patterns are stripped to get the base directory path.
     """
+    import gcsfs
     import pyarrow.dataset as ds
 
     clean_path = (
@@ -117,6 +116,7 @@ def sink_to_gcs(
     **kwargs
         Additional arguments passed to the underlying writer.
     """
+    import gcsfs
     import pyarrow.parquet as pq
 
     fs = gcsfs.GCSFileSystem(**storage_options)
@@ -161,6 +161,9 @@ def write_delta_to_gcs(
     partition_by
         Delta partition columns (applied at table creation).
     """
+    import gcsfs
+    from deltalake import write_deltalake
+
     clean_path = get_path_without_scheme(path)
 
     fs = gcsfs.GCSFileSystem(**storage_options)
@@ -205,6 +208,9 @@ def scan_delta_from_gcs(
     pl.LazyFrame
         A lazy frame backed by the Delta table's PyArrow dataset.
     """
+    import gcsfs
+    from deltalake import DeltaTable
+
     fs = gcsfs.GCSFileSystem(**storage_options)
     clean_path = get_path_without_scheme(resource_path)
 
diff --git a/shared/cloud_storage/utils.py b/shared/cloud_storage/utils.py
index 1e6f058df..ddd9b3a3b 100644
--- a/shared/cloud_storage/utils.py
+++ b/shared/cloud_storage/utils.py
@@ -7,8 +7,6 @@
 
 from typing import Any, Literal
 
-import boto3
-
 
 def normalize_delta_path(resource_path: str) -> str:
     """Normalize az:// paths to abfss:// for delta-rs compatibility.
@@ -42,6 +40,8 @@ def create_storage_options_from_boto_credentials(
     dict[str, Any]
         A storage options dictionary for Polars with explicit credentials.
     """
+    import boto3
+
     session = boto3.Session(profile_name=profile_name, region_name=region_name)
     credentials = session.get_credentials()
     frozen_creds = credentials.get_frozen_credentials()