Replace inspector with query to check if column exists in DB (#962)

* Replace inspector with query to check if column exists in DB

* Add test and updated check for query

* Fix linting'

* Delete orchestrator/config/settings.py

* Update test_migration_does_column_exist.py

* fix linting

---------

Co-authored-by: Peter Boers <peter.boers@surf.nl>

Author: Igoranze

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 4.0.3
+current_version = 4.0.4
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(rc(?P<build>\d+))?

orchestrator/__init__.py

@@ -13,7 +13,7 @@
 
 """This is the orchestrator workflow engine."""
 
-__version__ = "4.0.3"
+__version__ = "4.0.4"
 
 from orchestrator.app import OrchestratorCore
 from orchestrator.settings import app_settings

orchestrator/migrations/helpers.py

@@ -35,14 +35,20 @@ def has_table_column(table_name: str, column_name: str, conn: sa.engine.Connecti
     :param conn: SQLAlchemy database Connection
     :return: True if the column exists, False otherwise
     """
-    inspector = sa.inspect(conn.engine)
-    try:
-        columns = inspector.get_columns(table_name)
-        return any(col["name"] == column_name for col in columns)
-    except sa.exc.NoSuchTableError:
-        # On some migrations the table might not exist yet, so we catch the exception
-        logger.warning(f"Table {table_name} does not exist.")
-        return False
+    result = conn.execute(
+        sa.text(
+            """
+            SELECT column_name
+            FROM information_schema.columns
+            WHERE table_name = :table_name and column_name = :column_name
+            """
+        ),
+        {
+            "table_name": table_name,
+            "column_name": column_name,
+        },
+    )
+    return result.first() is not None
 
 
 def get_resource_type_id_by_name(conn: sa.engine.Connection, name: str) -> UUID:

test/unit_tests/db/test_migration_does_column_exist.py

@@ -0,0 +1,10 @@
+from orchestrator.db import db
+from orchestrator.migrations.helpers import has_table_column
+
+
+def test_select_from_table():
+    # Testing if Table Workflows exist with column is_task
+    # it should because the db.session depends on the session where all migrations are already run
+    session = db.session
+    result = has_table_column(table_name="workflows", column_name="is_task", conn=session)
+    assert result is True, "Column 'is_task' does not exist in 'workflows' table"