Add PENDING_COMPONENTS to show initializing applications

superduper/backends/base/queue.py

@@ -85,10 +85,6 @@ def __init__(self, uri: t.Optional[str]):
         self.uri: t.Optional[str] = uri
         self.queue: t.Dict = defaultdict(lambda: [])
 
-    @abstractmethod
-    def show_pending_create_events(self, type_id: str | None = None):
-        """Show pending events."""
-
     @abstractmethod
     def build_consumer(self, **kwargs):
         """Build a consumer instance."""
@@ -245,6 +241,7 @@ def consume_events(events, table: str, db=None):
 
         for event in table_events:
             event.execute(db)
+
         db.metadata.commit()
 
         # non table events

superduper/base/datalayer.py

@@ -149,7 +149,6 @@ def show(
         identifier: t.Optional[str] = None,
         version: t.Optional[int] = None,
         uuid: t.Optional[str] = None,
-        show_status: bool = False,
     ):
         """
         Show available functionality which has been added using ``self.add``.
@@ -173,29 +172,10 @@ def show(
             out = self.metadata.show_components()
             out = sorted(list(set([nt(**x) for x in out])))
             out = [x._asdict() for x in out]
-            if show_status:
-                out = [{**r, 'status': 'initialized'} for r in out]
-                out.extend(
-                    [
-                        {**r, 'status': 'pending'}
-                        for r in self.cluster.queue.show_pending_create_events(type_id)
-                    ]
-                )
             return out
 
         if identifier is None:
             out = self.metadata.show_components(type_id=type_id)
-            if show_status:
-                out = [
-                    {'identifier': x, 'type_id': type_id, 'status': 'initialized'}
-                    for x in out
-                ]
-                out.extend(
-                    [
-                        {'identifier': x, 'status': 'pending', 'type_id': type_id}
-                        for x in self.cluster.queue.show_pending_create_events(type_id)
-                    ]
-                )
             return sorted(out)
 
         if version is None:

superduper/rest/build.py

@@ -8,6 +8,10 @@
 import zipfile
 from contextlib import contextmanager
 
+from superduper.base.leaf import import_item
+from superduper.components.application import Application
+from superduper.misc.importing import import_object
+
 try:
     import magic
 except ImportError:
@@ -27,6 +31,9 @@
     from superduper.base.datalayer import Datalayer
 
 
+PENDING_COMPONENTS = set()
+
+
 class Tee:
     """A file-like object that writes to multiple outputs."""
 
@@ -192,11 +199,13 @@ def db_apply(
         id: str | None = 'test',
         db: 'Datalayer' = DatalayerDependency(),
     ):
+        cls_path = info['_builds'][info['_base'][1:]]['_path']
+        cls = import_object(cls_path)
+        type_id = cls.type_id
+        PENDING_COMPONENTS.add((type_id, info['identifier']))
         if '_variables' in info:
             info['_variables']['output_prefix'] = CFG.output_prefix
             info['_variables']['databackend'] = db.databackend.backend_name
-
-        # info = Document.decode(info, db=db).unpack()
         background_tasks.add_task(_process_db_apply, db, info, id)
         return {'status': 'ok'}
 
@@ -240,18 +249,28 @@ def db_show(
             r = db.metadata.get_component('application', application)
             return r['namespace']
         else:
-            return db.show(
+            out = db.show(
                 type_id=type_id,
                 identifier=identifier,
                 version=version,
-                show_status=show_status,
             )
+            if show_status:
+                assert version is None
+                if type_id is not None and identifier is None:
+                    out = [{'identifier': x, 'type_id': type_id} for x in out]
+                out = [{**r, 'status': 'initialized'} for r in out]
+                initialized = [(r['type_id'], r['identifier']) for r in out]
+                for pending_app in PENDING_COMPONENTS:
+                    if pending_app not in initialized:
+                        out.append({'type_id': pending_app[0], 'identifier': pending_app[1], 'status': 'pending'})
+            return out
 
     @app.add('/db/remove', method='post')
     def db_remove(
         type_id: str, identifier: str, db: 'Datalayer' = DatalayerDependency()
     ):
         db.remove(type_id=type_id, identifier=identifier, recursive=True, force=True)
+        PENDING_COMPONENTS.discard((type_id, identifier))
         return {'status': 'ok'}
 
     @app.add('/db/show_template', method='get')