Fix find_source_in_db crash on empty Sources table

astrodb_utils/sources.py

@@ -69,6 +69,12 @@ def find_source_in_db(
 
     source = source.strip()
 
+    # An empty Sources table has no matches by definition, and Database.query_region
+    # raises a KeyError (rather than returning zero rows) when queried against one.
+    if db.query(db.Sources).first() is None:
+        logger.debug(f"{source}: Sources table is empty, no matches possible.")
+        return []
+
     logger.debug(f"{source}: Searching for match in database. Use Simbad: {use_simbad}")
     db_name_matches = db.search_object(
         source,
@@ -129,8 +135,8 @@ def find_source_in_db(
             logger.debug(msg2)
             try:
                 db_name_matches = db.query_region(
-                simbad_skycoord, radius=search_radius, ra_col=ra_col_name, dec_col=dec_col_name
-            )
+                    simbad_skycoord, radius=search_radius, ra_col=ra_col_name, dec_col=dec_col_name
+                )
             except KeyError as e:
                 msg = (
                     f"Error querying database with coordinates from SIMBAD: {e}\n"

tests/test_sources.py

@@ -1,7 +1,9 @@
 import math
+import os
 
 import astropy.units as u
 import pytest
+from astrodbkit.astrodb import Database, create_database
 
 from astrodb_utils import (
     AstroDBError,
@@ -111,6 +113,29 @@ def test_find_source_in_db(db):
     )  # This is wrong and a result of fuzzy matching
 
 
+def test_find_source_in_db_empty_sources_table(tmp_path):
+    # Regression test for #265: the spatial fallback search in find_source_in_db
+    # used to crash with a pandas KeyError when the Sources table had zero rows.
+    template_schema_path = os.path.join("tests", "astrodb-template-db")
+    connection_string = f"sqlite:///{tmp_path}/empty_sources.sqlite"
+    create_database(connection_string, felis_schema=os.path.join(template_schema_path, "schema.yaml"))
+    empty_db = Database(connection_string)
+
+    try:
+        assert empty_db.query(empty_db.Sources).count() == 0
+
+        search_result = find_source_in_db(
+            empty_db,
+            "Test Source",
+            ra=10.0,
+            dec=20.0,
+            use_simbad=False,
+        )
+        assert search_result == []
+    finally:
+        empty_db.engine.dispose()
+
+
 def test_find_source_in_db_with_coords(db):
     ingest_source(
         db,