Fix SQLAlchemy 2.0 issues in create_database and document sql_query

astrodbkit/astrodb.py

@@ -163,7 +163,6 @@ def load_connection(connection_string, sqlite_foreign=True, base=None, connectio
     engine = create_engine(connection_string, connect_args=connection_arguments)
     if not base:
         base = declarative_base()
-    base.metadata.bind = engine
     Session = sessionmaker(bind=engine, query_cls=AstrodbQuery)  # pylint: disable=invalid-name
     session = Session()
 
@@ -234,7 +233,7 @@ def create_database(connection_string, drop_tables=False, felis_schema=None):
 
         # Drop tables, if requested
         if drop_tables:
-            base.metadata.drop_all()
+            base.metadata.drop_all(bind=engine)
 
         # Create the database
         metadata = MetaDataBuilder(schema).build()
@@ -243,7 +242,7 @@ def create_database(connection_string, drop_tables=False, felis_schema=None):
     else:
         session, base, engine = load_connection(connection_string, base=Base)
         if drop_tables:
-            base.metadata.drop_all()
+            base.metadata.drop_all(bind=engine)
         base.metadata.create_all(engine)  # this explicitly creates the database
 
     return session, base, engine
@@ -652,6 +651,8 @@ def search_string(self, value, fmt="table", fuzzy_search=True, verbose=True):
     def sql_query(self, query, fmt="default"):
         """
         Wrapper for a direct SQL query.
+        This is intended for read-only queries; changes made with statements such as
+        INSERT, UPDATE, or DELETE are not committed and will be rolled back.
 
         Parameters
         ----------

astrodbkit/tests/test_astrodb.py

@@ -72,6 +72,37 @@ def test_felis_db():
         os.remove(db_path)
 
 
+def test_drop_tables():
+    # Test that recreating a database with drop_tables=True clears existing data
+    db_path = "drop.db"
+
+    # Clear copy if file present
+    if os.path.exists(db_path):
+        os.remove(db_path)
+
+    connection_string = 'sqlite:///' + db_path
+    create_database(connection_string)
+    db = Database(connection_string)
+    with db.engine.begin() as conn:
+        conn.execute(db.Publications.insert().values(name='Ref 1'))
+    assert db.query(db.Publications).count() == 1
+    db.session.close()
+    db.engine.dispose()
+
+    # Recreate with drop_tables=True; tables should be dropped and recreated empty
+    create_database(connection_string, drop_tables=True)
+    db = Database(connection_string)
+    assert db.query(db.Publications).count() == 0
+    db.session.close()
+    db.engine.dispose()
+
+    # Felis branch should also accept drop_tables=True without error
+    create_database(connection_string, felis_schema="astrodbkit/schema_example_felis.yaml", drop_tables=True)
+
+    if os.path.exists(db_path):
+        os.remove(db_path)
+
+
 def test_add_data(db):
     # Load example data to the database
 

docs/index.rst

@@ -326,6 +326,9 @@ to pass direct SQL queries to the database for users who may wish to write their
     results = db.sql_query('select * from sources', fmt='astropy')
     print(results)
 
+Note that :py:meth:`~astrodbkit.astrodb.Database.sql_query` is meant for read-only queries.
+Changes made with statements like INSERT, UPDATE, or DELETE are not committed and will be rolled back.
+
 General Queries with Transformations
 ------------------------------------