Avoid adding/loading empty table, fixes #109

astrodbkit/astrodb.py

@@ -895,6 +895,10 @@ def add_table_data(self, data, table, fmt="csv"):
         else:
             raise RuntimeError(f"Unrecognized format {fmt}")
 
+        # Nothing to insert; SQLAlchemy would otherwise insert a single row of defaults
+        if df.empty:
+            return
+
         # Foreign key constraints will prevent inserts of missing sources,
         # but for clarity we'll check first and exit if there are missing sources
         if table != self._primary_table:
@@ -936,6 +940,11 @@ def load_table(self, table, directory, verbose=False):
         if os.path.exists(filename):
             with open(filename, "r", encoding="utf-8") as f:
                 data = json.load(f)
+                # SQLAlchemy treats an empty list as a single row of defaults, so skip empty files
+                if not data:
+                    if verbose:
+                        print(f"{table}.json is empty, skipping.")
+                    return
                 with self.engine.begin() as conn:
                     conn.execute(self.metadata.tables[table].insert().values(data))
         else:

astrodbkit/tests/test_astrodb.py

@@ -496,6 +496,29 @@ def test_load_database(db, db_dir):
             shutil.rmtree(file_path)
 
 
+def test_load_empty_table(tmp_path):
+    # Empty lookup JSON files (eg, []) should be skipped rather than inserting a blank row (issue #109)
+    connection_string = "sqlite:///" + str(tmp_path / "empty.db")
+    create_database(connection_string)
+    db = Database(connection_string)
+
+    with open(tmp_path / "Publications.json", "w") as f:
+        json.dump([], f)
+
+    db.load_table("Publications", str(tmp_path), verbose=True)
+    assert db.query(db.Publications).count() == 0
+
+    db.load_database(str(tmp_path))
+    assert db.query(db.Publications).count() == 0
+
+    # Empty DataFrame should also be a no-op
+    db.add_table_data(pd.DataFrame(columns=["name", "bibcode"]), table="Publications", fmt="pandas")
+    assert db.query(db.Publications).count() == 0
+
+    db.session.close()
+    db.engine.dispose()
+
+
 def test_copy_database_schema():
     connection_1 = 'sqlite:///' + DB_PATH
     connection_2 = 'sqlite:///second.db'