Add simple compute backend

Author: blythed

CHANGELOG.md

@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add recursive error propagation
 - Add reconnection support for Snowflake vector search
 - Add scanning for leaked passwords on the CI 
+- Simple detached-lifetime compute and vector-search implementations
 
 ### Bug fixes
 

superduper/backends/base/scheduler.py

@@ -10,7 +10,7 @@
 from superduper import CFG, logging
 from superduper.backends.base.backends import BaseBackend
 from superduper.base.base import Base
-from superduper.base.event import Create, CreateTable, PutComponent, Update
+from superduper.base.event import Create, CreateTable, Event, PutComponent, Update
 
 DependencyType = t.Union[t.Dict[str, str], t.Sequence[t.Dict[str, str]]]
 
@@ -142,7 +142,7 @@ def _consume_event_type(event_type, ids, table, db: 'Datalayer'):
 
 
 def cluster_events(
-    events: t.List[Base],
+    events: t.List[Event],
 ):
     """
     Cluster events into table, create and job events.
@@ -162,14 +162,14 @@ def cluster_events(
         elif isinstance(event, (Update, Create)):
             create_events.append(event)
         elif isinstance(event, Job):
-            job_events.append(event)
+            job_events.append(event)  # type: ignore[arg-type]
         elif isinstance(event, PutComponent):
             put_events.append(event)
     return table_events, create_events, put_events, job_events
 
 
 def consume_events(
-    events: t.List[Base],
+    events: t.List[Event],
     table: str,
     db: 'Datalayer',
     batch_size: int | None = None,

superduper/backends/local/scheduler.py

@@ -7,6 +7,7 @@
     consume_events,
 )
 from superduper.base import Base
+from superduper.base.event import Event
 from superduper.components.cdc import CDC
 from superduper.misc.importing import isreallyinstance
 
@@ -73,7 +74,7 @@ def initialize(self):
                 with self.lock:
                     self.Q[component, identifier] = []
 
-    def publish(self, events: t.List[Base]):
+    def publish(self, events: t.List[Event]):
         """
         Publish events to local queue.
 

superduper/backends/simple/__init__.py

@@ -1,3 +1,3 @@
 from .cluster import SimpleCluster as Cluster
 
-__all__ = ['Cluster']
+__all__ = ['Cluster']

superduper/backends/simple/cluster.py

@@ -1,15 +1,19 @@
+import os
 import typing as t
 
 import click
-import os
 
-from superduper import logging
+from superduper import logging, superduper
 from superduper.backends.base.cluster import Cluster
 from superduper.backends.local.cdc import LocalCDCBackend
-from superduper.backends.simple.compute import SimpleComputeClient
 from superduper.backends.local.crontab import LocalCrontabBackend
-from superduper.backends.simple.scheduler import SimpleScheduler
 from superduper.backends.local.vector_search import LocalVectorSearchBackend
+from superduper.backends.simple.compute import SimpleComputeBackend, SimpleComputeClient
+from superduper.backends.simple.scheduler import SimpleScheduler
+from superduper.backends.simple.vector_search import (
+    SimpleVectorSearch,
+    SimpleVectorSearchClient,
+)
 from superduper.misc.importing import load_plugin
 
 
@@ -24,15 +28,55 @@ class SimpleCluster(Cluster):
     :param crontab: The crontab backend.
     """
 
+    @classmethod
+    def build(cls, CFG, **kwargs):
+        """Build the local cluster."""
+        return SimpleCluster(
+            scheduler=SimpleScheduler(),
+            compute=SimpleComputeClient(),
+            vector_search=SimpleVectorSearchClient(),
+            cdc=LocalCDCBackend(),
+            crontab=LocalCrontabBackend(),
+        )
+
+    def drop(self, force: bool = False):
+        """Drop the cluster.
+
+        :param force: Force drop the cluster.
+        """
+        if not force:
+            if not click.confirm(
+                "Are you sure you want to drop the cluster? ",
+                default=False,
+            ):
+                logging.warn("Aborting...")
+
+            self.vector_search.drop(force=True)
+            self.compute.drop(force=True)
+
+
+class SimpleClusterBackend(Cluster):
+    """Simple cluster for running infra on a single machine.
+
+    :param compute: The compute backend.
+    :param cache: The cache backend.
+    :param scheduler: The scheduler backend.
+    :param vector_search: The vector search backend.
+    :param cdc: The change data capture backend
+    :param crontab: The crontab backend.
+    """
+
     @classmethod
     def build(cls, CFG, **kwargs):
         """Build the local cluster."""
         searcher_impl = load_plugin(CFG.vector_search_engine).VectorSearcher
 
         return SimpleCluster(
             scheduler=SimpleScheduler(),
-            compute=SimpleComputeClient(),
-            vector_search=LocalVectorSearchBackend(searcher_impl=searcher_impl),
+            compute=SimpleComputeBackend(),
+            vector_search=SimpleVectorSearch(
+                backend=LocalVectorSearchBackend(searcher_impl=searcher_impl)
+            ),
             cdc=LocalCDCBackend(),
             crontab=LocalCrontabBackend(),
         )
@@ -47,4 +91,30 @@ def drop(self, force: bool = False):
                 "Are you sure you want to drop the cluster? ",
                 default=False,
             ):
-                logging.warn("Aborting...")
+                logging.warn("Aborting...")
+
+            self.vector_search.drop()
+            self.compute.drop()
+
+
+if __name__ == "__main__":
+
+    import uvicorn
+    from fastapi import FastAPI
+
+    from superduper import CFG
+
+    db = superduper(cluster_engine='local')
+
+    app = FastAPI()
+
+    cluster = SimpleClusterBackend.build(CFG=CFG)
+
+    cluster.compute.build(app=app)
+    cluster.vector_search.build(app=app)
+
+    db.cluster = cluster
+
+    cluster.db = db
+
+    uvicorn.run(app, host='localhost', port=8001)