Add changes necessary for services

Author: blythed

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Move from in-line encoding to schema-based encoding with `Leaf._fields`
 - No need to define `_fields`
 - Use databackend to perform metadata duties
+- Add `db.create` and `db.insert` instead of `auto_schema`
 
 #### New Features & Functionality
 

plugins/ibis/plugin_test/config.yaml

@@ -1,4 +1,3 @@
-artifact_store: null
 data_backend: sqlite://
 auto_schema: false
 force_apply: true

plugins/ibis/plugin_test/test_query.py

@@ -2,7 +2,9 @@
 
 import numpy as np
 import pytest
+from superduper.base.base import Base
 from superduper.base.document import Document
+from superduper.components.listener import Listener
 from superduper.components.table import Table
 
 
@@ -83,8 +85,12 @@ def test_filter(db):
     assert len(r) == uq[1][0]
 
 
+class documents(Base):
+    this: 'str'
+
+
 def test_select_using_ids(db):
-    db.cfg.auto_schema = True
+    db.create(documents)
 
     table = db["documents"]
     table.insert([{"this": f"is a test {i}", "id": str(i)} for i in range(4)])
@@ -102,12 +108,17 @@ def test_select_using_ids_of_outputs(db):
     def my_func(x):
         return x + ' ' + x
 
-    db.cfg.auto_schema = True
+    db.create(documents)
 
     table = db["documents"]
     table.insert([{"this": f"is a test {i}", "id": str(i)} for i in range(4)])
 
-    listener = my_func.to_listener(key='this', select=db['documents'].select())
+    listener = Listener(
+        'test',
+        model=my_func,
+        key='this',
+        select=db['documents'].select(),
+    )
     db.apply(listener)
 
     q1 = db[listener.outputs].select()

plugins/ibis/plugin_test/test_sanity.py

@@ -7,33 +7,31 @@ class New(Base):
 
 def test(db):
 
-    db.create(New)
-
     data = [
-        New('testa', a='a').dict(path=False),
-        New('testb', a='b').dict(path=False),
-        New('testc', a='c').dict(path=False),
+        New(a='a'),
+        New(a='b'),
+        New(a='c').dict(path=False),
     ]
 
-    db['New'].insert(data)
+    db.insert(data)
 
     loaded = db['New'].execute()
 
     assert len(loaded) == 3
 
     db['New'].update({'a': 'a'}, 'a', 'a2')
 
-    r = db['New'].get(identifier='testa')
+    r = db['New'].get(a='a2')
 
     assert r['a'] == 'a2'
 
     r['a'] = 'a3'
 
-    db['New'].replace({'identifier': 'testa'}, r)
+    db['New'].replace({'a': 'a2'}, r)
 
-    assert db['New'].get(identifier='testa')['a'] == 'a3'
+    assert db['New'].get(a='a3')['a'] == 'a3'
 
-    db['New'].delete({'identifier': 'testa'})
+    db['New'].delete({'a': 'a3'})
     assert len(db['New'].execute()) == 2
 
-    assert db['New'].get(identifier='testa') is None
+    assert db['New'].get(a='a3') is None

plugins/ibis/superduper_ibis/data_backend.py

@@ -133,15 +133,15 @@ def check_output_dest(self, predict_id) -> bool:
         except (NoSuchTableError, ibis.IbisError):
             return False
 
-    def create_table_and_schema(self, identifier: str, schema: Schema):
+    def create_table_and_schema(self, identifier: str, schema: Schema, primary_id: str):
         """Create a schema in the data-backend.
 
         :param identifier: The identifier of the table.
         :param mapping: The mapping of the schema.
         """
         mapping = convert_schema_to_fields(schema)
-        if "id" not in mapping:
-            mapping["id"] = "string"
+        if primary_id not in mapping:
+            mapping[primary_id] = "string"
         try:
             mapping = self.db_helper.process_schema_types(mapping)
             t = self._create_table_with_retry(identifier, schema=ibis.schema(mapping))

plugins/mongodb/plugin_test/config.yaml

@@ -1,4 +1,3 @@
-artifact_store: null
 data_backend: mongomock://test_db
 auto_schema: false
 force_apply: true

plugins/mongodb/superduper_mongodb/data_backend.py

@@ -111,7 +111,7 @@ def check_output_dest(self, predict_id) -> bool:
         """
         return self._database[f"{CFG.output_prefix}{predict_id}"].find_one() is not None
 
-    def create_table_and_schema(self, identifier: str, schema: Schema):
+    def create_table_and_schema(self, identifier: str, schema: Schema, primary_id: str):
         """Create a table and schema in the data backend.
 
         :param identifier: The identifier for the table

plugins/openai/superduper_openai/model.py

@@ -3,8 +3,8 @@
 import json
 import os
 import typing as t
-
 from functools import lru_cache as cache
+
 import numpy
 import requests
 import tqdm

pyproject.toml

@@ -136,7 +136,8 @@ ignore = [
   "D401",
   "D102",
   "E402",
-  "F403"
+  "F403",
+  "F401", # Removing lines of code
 ]
 exclude = ["templates", "superduper/templates"]
 

superduper/backends/base/cluster.py

@@ -83,7 +83,7 @@ def load_custom_plugins(self):
         from superduper import logging
 
         if 'Plugin' in self.db.show('Table'):
-            logging.info(f"Found custom plugins - loading...")
+            logging.info("Found custom plugins - loading...")
             for plugin in self.db.show('Plugin'):
                 logging.info(f"Loading plugin: {plugin}")
                 plugin = self.db.load('Plugin', plugin)

superduper/backends/base/data_backend.py

@@ -75,11 +75,14 @@ def build_artifact_store(self):
         pass
 
     @abstractmethod
-    def create_table_and_schema(self, identifier: str, schema: 'Schema'):
+    def create_table_and_schema(
+        self, identifier: str, schema: 'Schema', primary_id: str
+    ):
         """Create a schema in the data-backend.
 
         :param identifier: The identifier of the schema.
         :param schema: The schema to create.
+        :param primary_id: The primary id of the schema.
         """
 
     @abstractmethod

superduper/backends/base/metadata.py

@@ -4,9 +4,10 @@
 import uuid
 
 from superduper import logging
-from superduper.base.exceptions import DatabackendError
 from superduper.base.base import Base
+from superduper.base.exceptions import DatabackendError
 from superduper.components.cdc import CDC
+from superduper.components.component import Component
 from superduper.components.schema import Schema
 from superduper.components.table import Table
 from superduper.misc.importing import import_object
@@ -113,7 +114,7 @@ def __init__(self, db):
             ).encode(),
             ('Table', 'Job'): Table(
                 identifier='Job',
-                primary_id='uuid',
+                primary_id='job_id',
                 uuid='jkl',
                 component=True,
                 path='superduper.backends.base.metadata.Job',
@@ -127,7 +128,12 @@ def __init__(self, db):
     def init(self):
         """Initialize the metadata store."""
         for cls in metaclasses.values():
-            self.db.databackend.create_table_and_schema(cls.__name__, cls.class_schema)
+            preset = self.preset_components.get(('Table', cls.__name__))
+            self.db.databackend.create_table_and_schema(
+                cls.__name__,
+                cls.class_schema,
+                primary_id=preset['primary_id'],
+            )
 
     def get_schema(self, table: str):
         """Get the schema of a table.
@@ -162,10 +168,18 @@ def create(self, cls: t.Type[Base]):
             r = self.db['Table'].get(identifier=cls.__name__)
         except DatabackendError as e:
             if 'not found' in str(e):
-                self.db.databackend.create_table_and_schema('Table', Table.class_schema)
-                t = Table('Table', path='superduper.components.table.Table', primary_id='uuid', component=True)
+                self.db.databackend.create_table_and_schema(
+                    'Table', Table.class_schema, 'uuid'
+                )
+                t = Table(
+                    'Table',
+                    path='superduper.components.table.Table',
+                    primary_id='uuid',
+                    component=True,
+                )
+
                 r = self.db['Table'].insert(
-                    [t.dict(schema=True, path=False)], 
+                    [t.dict(schema=True, path=False)],
                 )
             else:
                 raise e
@@ -175,8 +189,16 @@ def create(self, cls: t.Type[Base]):
                 f'{cls.__name__} already exists in metadata with data: {r}'
             )
 
-        self.db.databackend.create_table_and_schema(cls.__name__, cls.class_schema)
-        t = Table(identifier=cls.__name__, path=f'{cls.__module__}.{cls.__name__}', primary_id='uuid', component=True)
+        pid = 'uuid' if issubclass(cls, Component) else self.db.databackend.id_field
+        self.db.databackend.create_table_and_schema(
+            cls.__name__, cls.class_schema, primary_id=pid
+        )
+        t = Table(
+            identifier=cls.__name__,
+            path=f'{cls.__module__}.{cls.__name__}',
+            primary_id=pid,
+            component=True,
+        )
         self.db['Table'].insert([t.dict(path=False)])
         return t
 
@@ -198,9 +220,17 @@ def create_entry(
         :param component: component name
         :param raw: whether to insert raw data
         """
+        path = None
         if component is None:
             path = info.pop('_path')
             component = path.rsplit('.', 1)[1]
+
+        metadata = self.db['Table'].get(component)
+        if metadata is None:
+            assert path is not None
+            cls = import_object(path)
+            self.create(cls)
+
         self.db[component].insert([info], raw=raw)
 
     def create_component(self, info: t.Dict):
@@ -339,10 +369,14 @@ def show_jobs(self, component: str, identifier: str):
         :param component: type of component
         :param identifier: identifier of component
         """
-        return self.db['Job'].filter(
-            self.db['Job']['component'] == component,
-            self.db['Job']['identifier'] == identifier,
-        ).distinct('job_id')
+        return (
+            self.db['Job']
+            .filter(
+                self.db['Job']['component'] == component,
+                self.db['Job']['identifier'] == identifier,
+            )
+            .distinct('job_id')
+        )
 
     def show_components(self, component: str | None = None):
         """
@@ -492,6 +526,10 @@ def get_component_by_uuid(self, component: str, uuid: str):
         if uuid in self.preset_uuids:
             return self.preset_uuids[uuid]
         r = self.db[component].get(uuid=uuid, raw=True)
+        if r is None:
+            raise NonExistentMetadataError(
+                f'Object {uuid} does not exist in metadata for {component}'
+            )
         path = self.db['Table'].get(identifier=component)['path']
         r['_path'] = path
         return r

superduper/backends/base/query.py

@@ -18,9 +18,9 @@
 
 from superduper import CFG, logging
 from superduper.backends.base.metadata import NonExistentMetadataError
+from superduper.base.base import Base
 from superduper.base.constant import KEY_BLOBS, KEY_BUILDS, KEY_FILES, KEY_PATH
 from superduper.base.document import Document, _unpack
-from superduper.base.base import Base
 
 if t.TYPE_CHECKING:
     from superduper.base.datalayer import Datalayer

superduper/base/apply.py

@@ -301,11 +301,9 @@ def replace_existing(x):
             # during the `.map` to the children
             # serializer.map...
             # this means replacing components with references
-            serialized = object.dict().update(serialized)
+            serialized = object.dict(metadata=False).update(serialized)
 
             # this is necessary to prevent inconsistencies
-            # this takes the difference between
-            # the current and
             serialized = serialized.update(this_diff).encode(keep_schema=False)
 
             # assign/ increment the version since
@@ -339,7 +337,7 @@ def replace_existing(x):
             # update the existing component with the change
             # data from the applied component
             serialized = (
-                current.dict()
+                current.dict(metadata=False)
                 .update(serialized)
                 .update(this_diff)
                 .encode(keep_schema=False)
@@ -354,7 +352,11 @@ def replace_existing(x):
         )
         serialized['version'] = 0
 
-        serialized = object.dict().update(serialized)
+        # TODO this was set to empty (no metadata specified) previously
+        # Do we even need this?
+        # The metadata components could "simple" sit quietly inside the component
+        # and be applied when the component is applied
+        serialized = object.dict(metadata=False).update(serialized)
 
         # if the metadata includes components, which
         # need to be applied, do that now