[executors] fix: init super classes for Bq and SQl executors

* Provide query title when executor fails to run the query
* Provide INFO logging messages for bq executor

Author: AndreyMarkinPPC

libs/executors/garf_executors/__init__.py

@@ -57,4 +57,4 @@ def setup_executor(
   'ApiExecutionContext',
 ]
 
-__version__ = '0.2.0'
+__version__ = '0.2.1'

libs/executors/garf_executors/bq_executor.py

@@ -30,6 +30,7 @@
 
 from garf_core import query_editor, report
 from google.cloud import exceptions as google_cloud_exceptions
+from opentelemetry import trace
 
 from garf_executors import exceptions, execution_context, executor
 from garf_executors.telemetry import tracer
@@ -68,6 +69,7 @@ def __init__(
       )
     self.project_id = project_id
     self.location = location
+    super().__init__()
 
   @property
   def client(self) -> bigquery.Client:
@@ -93,41 +95,47 @@ def execute(
     Returns:
       Report with data if query returns some data otherwise empty Report.
     """
+    span = trace.get_current_span()
+    logger.info('Executing script: %s', title)
     query_text = self.replace_params_template(query, context.query_parameters)
     self.create_datasets(context.query_parameters.macro)
     job = self.client.query(query_text)
     try:
       result = job.result()
+    except google_cloud_exceptions.GoogleCloudError as e:
+      raise BigQueryExecutorError(
+        f'Failed to execute query {title}: Reason: {e}'
+      ) from e
       logger.debug('%s launched successfully', title)
-      if result.total_rows:
-        results = report.GarfReport.from_pandas(result.to_dataframe())
+    if result.total_rows:
+      results = report.GarfReport.from_pandas(result.to_dataframe())
+    else:
+      results = report.GarfReport()
+    if context.writer and results:
+      writer_clients = context.writer_clients
+      if not writer_clients:
+        logger.warning('No writers configured, skipping write operation')
       else:
-        results = report.GarfReport()
-      if context.writer and results:
-        writer_clients = context.writer_clients
-        if not writer_clients:
-          logger.warning('No writers configured, skipping write operation')
-        else:
-          writing_results = []
-          for writer_client in writer_clients:
-            logger.debug(
-              'Start writing data for query %s via %s writer',
-              title,
-              type(writer_client),
-            )
-            writing_result = writer_client.write(results, title)
-            logger.debug(
-              'Finish writing data for query %s via %s writer',
-              title,
-              type(writer_client),
-            )
-            writing_results.append(writing_result)
-          logger.info('%s executed successfully', title)
-          # Return the last writer's result for backward compatibility
-          return writing_results[-1] if writing_results else None
-      return results
-    except google_cloud_exceptions.GoogleCloudError as e:
-      raise BigQueryExecutorError(e) from e
+        writing_results = []
+        for writer_client in writer_clients:
+          logger.debug(
+            'Start writing data for query %s via %s writer',
+            title,
+            type(writer_client),
+          )
+          writing_result = writer_client.write(results, title)
+          logger.debug(
+            'Finish writing data for query %s via %s writer',
+            title,
+            type(writer_client),
+          )
+          writing_results.append(writing_result)
+        # Return the last writer's result for backward compatibility
+        logger.info('%s executed successfully', title)
+        return writing_results[-1] if writing_results else None
+    logger.info('%s executed successfully', title)
+    span.set_attribute('execute.num_results', len(results))
+    return results
 
   @tracer.start_as_current_span('bq.create_datasets')
   def create_datasets(self, macros: dict | None) -> None:

libs/executors/garf_executors/sql_executor.py

@@ -57,6 +57,7 @@ def __init__(self, engine: sqlalchemy.engine.base.Engine) -> None:
         engine: Initialized Engine object to operated on a given database.
     """
     self.engine = engine
+    super().__init__()
 
   @classmethod
   def from_connection_string(
@@ -89,12 +90,17 @@ def execute(
       Report with data if query returns some data otherwise empty Report.
     """
     span = trace.get_current_span()
-    logging.info('Executing script: %s', title)
+    logger.info('Executing script: %s', title)
     query_text = self.replace_params_template(query, context.query_parameters)
     with self.engine.begin() as conn:
       if re.findall(r'(create|update) ', query_text.lower()):
-        conn.connection.executescript(query_text)
-        results = report.GarfReport()
+        try:
+          conn.connection.executescript(query_text)
+          results = report.GarfReport()
+        except Exception as e:
+          raise SqlAlchemyQueryExecutorError(
+            f'Failed to execute query {title}: Reason: {e}'
+          ) from e
       else:
         temp_table_name = f'temp_{uuid.uuid4().hex}'
         query_text = f'CREATE TABLE {temp_table_name} AS {query_text}'
@@ -103,6 +109,10 @@ def execute(
           results = report.GarfReport.from_pandas(
             pd.read_sql(f'SELECT * FROM {temp_table_name}', conn)
           )
+        except Exception as e:
+          raise SqlAlchemyQueryExecutorError(
+            f'Failed to execute query {title}: Reason: {e}'
+          ) from e
         finally:
           conn.connection.execute(f'DROP TABLE {temp_table_name}')
       if context.writer and results: