Fix: Allow empty files in models folder and refetch lineage on model file change (#746)

Author: mykalmax

setup.py

@@ -98,7 +98,7 @@
         ],
         "web": [
             "fastapi==0.95.0",
-            "hyperscript==0.0.1",
+            "watchfiles==0.19.0",
             "pyarrow==11.0.0",
             "uvicorn==0.21.1",
         ],

sqlmesh/core/loader.py

@@ -193,6 +193,9 @@ def _load_sql_models(
             cache = SqlMeshLoader._Cache(self, context_path)
 
             for path in self._glob_paths(context_path / c.MODELS, config=config, extension=".sql"):
+                if not os.path.getsize(path):
+                    continue
+
                 self._track_file(path)
 
                 def _load() -> Model:
@@ -234,6 +237,9 @@ def _load_python_models(self) -> UniqueKeyDict[str, Model]:
 
         for context_path, config in self._context.configs.items():
             for path in self._glob_paths(context_path / c.MODELS, config=config, extension=".py"):
+                if not os.path.getsize(path):
+                    continue
+
                 self._track_file(path)
                 self._import_python_file(path, context_path)
                 new = registry.keys() - registered

web/client/src/library/components/editor/EditorPreview.tsx

@@ -12,9 +12,8 @@ import { ViewColumnsIcon } from '@heroicons/react/24/solid'
 import { Button } from '@components/button/Button'
 import { EnumVariant } from '~/types/enum'
 import { useStoreContext } from '@context/context'
-import { apiCancelLineage, useApiModelLineage } from '@api/index'
+import { useApiModelLineage } from '@api/index'
 import Graph from '@components/graph/Graph'
-import { useQueryClient } from '@tanstack/react-query'
 import Loading from '@components/loading/Loading'
 
 export const EnumEditorPreviewTabs = {
@@ -40,6 +39,7 @@ export default function EditorPreview({
   const previewTable = useStoreEditor(s => s.previewTable)
 
   const [activeTabIndex, setActiveTabIndex] = useState(-1)
+  const [modelName, setModelName] = useState<string | undefined>()
 
   const tabs = useMemo(() => {
     if (tab.file.isLocal)
@@ -58,6 +58,7 @@ export default function EditorPreview({
 
     return [EnumEditorPreviewTabs.Console]
   }, [tab])
+
   const [headers, data] = useMemo(
     () =>
       previewTable == null
@@ -88,6 +89,10 @@ export default function EditorPreview({
     setActiveTabIndex(tab.file.isSQLMeshModel ? 3 : -1)
   }, [previewTable, previewQuery, previewConsole, tab])
 
+  useEffect(() => {
+    setModelName(models.get(tab.file.path)?.name)
+  }, [models, tab.file])
+
   const table = useReactTable({
     data,
     columns,
@@ -106,8 +111,6 @@ export default function EditorPreview({
     return tabName === EnumEditorPreviewTabs.Query && isNil(previewQuery)
   }
 
-  const model = models.get(tab.file.path)
-
   return (
     <div
       className={clsx(
@@ -262,12 +265,13 @@ export default function EditorPreview({
               'w-full h-full ring-white ring-opacity-60 ring-offset-2 ring-offset-blue-400 focus:outline-none focus:ring-2 p-2',
             )}
           >
-            {model == null ? (
+            {modelName == null ? (
               <div>Model Does Not Exist</div>
             ) : (
               <EditorPreviewLineage
-                key={model.name}
-                model={model.name}
+                key={modelName}
+                model={modelName}
+                tab={tab}
               />
             )}
           </Tab.Panel>
@@ -277,44 +281,42 @@ export default function EditorPreview({
   )
 }
 
-function EditorPreviewLineage({ model }: { model: string }): JSX.Element {
-  const client = useQueryClient()
-
-  const { data: dag, refetch: getModelLineage } = useApiModelLineage(model)
+function EditorPreviewLineage({
+  model,
+  tab,
+}: {
+  model: string
+  tab: EditorTab
+}): JSX.Element {
+  const { data: lineage, refetch: getModelLineage } = useApiModelLineage(model)
 
   const previewLineage = useStoreEditor(s => s.previewLineage)
   const setPreviewLineage = useStoreEditor(s => s.setPreviewLineage)
 
   const debouncedGetModelLineage = useCallback(
     debounceAsync(getModelLineage, 1000, true),
-    [model],
+    [model, tab.file.path, tab.file.fingerprint],
   )
 
   useEffect(() => {
     void debouncedGetModelLineage()
-
-    return () => {
-      debouncedGetModelLineage.cancel()
-
-      apiCancelLineage(client)
-    }
-  }, [model])
+  }, [debouncedGetModelLineage])
 
   useEffect(() => {
-    if (dag == null) {
+    if (lineage == null) {
       setPreviewLineage(undefined)
     } else {
       setPreviewLineage(
-        Object.keys(dag).reduce((acc: Record<string, Lineage>, key) => {
+        Object.keys(lineage).reduce((acc: Record<string, Lineage>, key) => {
           acc[key] = {
-            models: dag[key] ?? [],
+            models: lineage[key] ?? [],
           }
 
           return acc
         }, {}),
       )
     }
-  }, [dag])
+  }, [lineage])
 
   return previewLineage == null ? (
     <div className="w-full h-full flex items-center justify-center bg-primary-10">

web/client/src/library/components/ide/IDE.tsx

@@ -62,6 +62,7 @@ export function IDE(): JSX.Element {
 
   useEffect(() => {
     const unsubscribeTasks = subscribe('tasks', updateTasks)
+    const unsubscribeModels = subscribe('models', setModels)
 
     void debouncedGetEnvironemnts()
     void debouncedGetFiles()
@@ -77,6 +78,7 @@ export function IDE(): JSX.Element {
       apiCancelGetEnvironments(client)
 
       unsubscribeTasks?.()
+      unsubscribeModels?.()
     }
   }, [])
 

web/client/src/models/file.ts

@@ -52,6 +52,10 @@ export class ModelFile extends ModelArtifact<InitialFile> {
     return this.type === 'model'
   }
 
+  get fingerprint(): string {
+    return this._content
+  }
+
   updateContent(newContent: string = ''): void {
     this._content = newContent
     this.content = newContent

web/server/api/endpoints/files.py

@@ -87,7 +87,7 @@ def walk_path(
 def get_file(
     path: str = Depends(validate_path),
     settings: Settings = Depends(get_settings),
-    path_mapping: t.Dict[str, t.Any] = Depends(get_path_mapping),
+    path_mapping: t.Dict[Path, models.FileType] = Depends(get_path_mapping),
 ) -> models.File:
     """Get a file, including its contents."""
     try:
@@ -96,7 +96,7 @@ def get_file(
     except FileNotFoundError:
         raise HTTPException(status_code=HTTP_404_NOT_FOUND)
     return models.File(
-        name=os.path.basename(path), path=path, content=content, type=path_mapping.get(path)
+        name=os.path.basename(path), path=path, content=content, type=path_mapping.get(Path(path))
     )
 
 
@@ -107,6 +107,7 @@ async def write_file(
     path: str = Depends(validate_path),
     settings: Settings = Depends(get_settings),
     context: Context = Depends(get_context),
+    path_mapping: t.Dict[Path, models.FileType] = Depends(get_path_mapping),
 ) -> models.File:
     """Create, update, or rename a file."""
     path_or_new_path = path
@@ -118,7 +119,10 @@ async def write_file(
 
     content = (settings.project_path / path_or_new_path).read_text()
     return models.File(
-        name=os.path.basename(path_or_new_path), path=path_or_new_path, content=content
+        name=os.path.basename(path_or_new_path),
+        path=path_or_new_path,
+        content=content,
+        type=path_mapping.get(Path(path)),
     )
 
 

web/server/api/endpoints/plan.py

@@ -78,7 +78,7 @@ async def run_plan(
                     [to_ds(t) for t in make_inclusive(start, end)]
                     for start, end in interval.merged_intervals
                 ][0],
-                batches=tasks[interval.snapshot_name],
+                batches=tasks.get(interval.snapshot_name, 0),
             )
             for interval in plan.missing_intervals
         ]

web/server/main.py

@@ -7,6 +7,7 @@
 
 from web.server.api.endpoints import api_router
 from web.server.console import ApiConsole
+from web.server.watcher import watch_project
 
 app = FastAPI()
 api_console = ApiConsole()
@@ -26,11 +27,13 @@ async def dispatch() -> None:
 
     app.state.console_listeners = []
     app.state.dispatch_task = asyncio.create_task(dispatch())
+    app.state.watch_task = asyncio.create_task(watch_project(api_console.queue))
 
 
 @app.on_event("shutdown")
 def shutdown_event() -> None:
     app.state.dispatch_task.cancel()
+    app.state.watch_task.cancel()
 
 
 @app.get("/health")

web/server/watcher.py

@@ -0,0 +1,26 @@
+import asyncio
+import json
+
+from watchfiles import DefaultFilter, awatch
+
+from sqlmesh.core import constants as c
+from web.server.api.endpoints.models import get_models
+from web.server.settings import get_loaded_context, get_settings
+from web.server.sse import Event
+
+
+async def watch_project(queue: asyncio.Queue) -> None:
+    settings = get_settings()
+    context = await get_loaded_context(settings)
+
+    async for _ in awatch(
+        (context.path / c.MODELS).resolve(),
+        watch_filter=DefaultFilter(
+            ignore_entity_patterns=context.config.ignore_patterns if context else c.IGNORE_PATTERNS
+        ),
+    ):
+        context.load()
+
+        queue.put_nowait(
+            Event(event="models", data=json.dumps([model.dict() for model in get_models(context)]))
+        )