[executors] feat: unify bq and sqldb executors

* Add support for bq and sqldb executors in CLI and server
* Add `create_executor` package level function.
* BQ and SqlAlchemy executors returns GarfReport instead of DataFrame;
If writer is provided the results are written to specified writer
destination, otherwise results of execution are returned back to caller.
* Add ExecutionContext class for all executors that contains writer,
  query_, fetcher_ and writer_parameters. ApiExecutionContext contains
  predefined 'console' writer.

Author: AndreyMarkinPPC

libs/garf_executors/garf_executors/__init__.py

@@ -15,9 +15,32 @@
 
 from __future__ import annotations
 
+from garf_executors import bq_executor, exceptions, sql_executor
 from garf_executors.api_executor import ApiExecutionContext, ApiQueryExecutor
 from garf_executors.fetchers import FETCHERS
 
+
+def setup_executor(source: str, fetcher_parameters: dict[str, str]):
+  """Initializes executors based on a source and parameters."""
+  if source not in ('bq', 'sqldb') and not (
+    concrete_api_fetcher := FETCHERS.get(source)
+  ):
+    raise exceptions.GarfExecutorError(f'Source {source} is not available.')
+  if source == 'bq':
+    query_executor = bq_executor.BigQueryExecutor(**fetcher_parameters)
+  elif source == 'sqldb':
+    query_executor = (
+      sql_executor.SqlAlchemyQueryExecutor.from_connection_string(
+        fetcher_parameters.get('connection_string')
+      )
+    )
+  else:
+    query_executor = ApiQueryExecutor(
+      concrete_api_fetcher(**fetcher_parameters)
+    )
+  return query_executor
+
+
 __all__ = [
   'FETCHERS',
   'ApiQueryExecutor',

libs/garf_executors/garf_executors/api_executor.py

@@ -22,45 +22,16 @@
 
 import logging
 
-import pydantic
-
-from garf_core import query_editor, report_fetcher
-from garf_executors import exceptions
-from garf_io import writer
-from garf_io.writers import abs_writer
+from garf_core import report_fetcher
+from garf_executors import exceptions, execution_context
 
 logger = logging.getLogger(__name__)
 
 
-class ApiExecutionContext(pydantic.BaseModel):
-  """Common context for executing one or more queries.
-
-  Attributes:
-    query_parameters: Parameters to dynamically change query text.
-    fetcher_parameters: Parameters to specify fetching setup.
-    writer: Type of writer to use.
-    writer_parameters: Optional parameters to setup writer.
-  """
+class ApiExecutionContext(execution_context.ExecutionContext):
+  """Common context for executing one or more queries."""
 
-  query_parameters: query_editor.GarfQueryParameters | None = None
-  fetcher_parameters: dict[str, str] | None = None
   writer: str = 'console'
-  writer_parameters: dict[str, str] | None = None
-
-  def model_post_init(self, __context__) -> None:
-    if self.fetcher_parameters is None:
-      self.fetcher_parameters = {}
-    if self.writer_parameters is None:
-      self.writer_parameters = {}
-
-  @property
-  def writer_client(self) -> abs_writer.AbsWriter:
-    writer_client = writer.create_writer(self.writer, **self.writer_parameters)
-    if self.writer == 'bq':
-      _ = writer_client.create_or_get_dataset()
-    if self.writer == 'sheet':
-      writer_client.init_client()
-    return writer_client
 
 
 class ApiQueryExecutor:
@@ -79,15 +50,22 @@ def __init__(self, fetcher: report_fetcher.ApiReportFetcher) -> None:
     self.fetcher = fetcher
 
   async def aexecute(
-    self, query: str, context: ApiExecutionContext, **kwargs: str
+    self,
+    query: str,
+    title: str,
+    context: ApiExecutionContext,
   ) -> str:
-    """Reads query, extract results and stores them in a specified location.
+    """Performs query execution asynchronously.
 
     Args:
       query: Location of the query.
+      title: Name of the query.
       context: Query execution context.
+
+    Returns:
+      Result of writing the report.
     """
-    await self.execute(query, context, **kwargs)
+    return await self.execute(query, context, title, context)
 
   def execute(
     self,

libs/garf_executors/garf_executors/bq_executor.py

@@ -25,16 +25,16 @@
 
 import logging
 
-import pandas as pd
 from google.cloud import exceptions as google_cloud_exceptions
 
-from garf_core import query_editor
+from garf_core import query_editor, report
+from garf_executors import exceptions, execution_context
 
 logger = logging.getLogger(__name__)
 
 
-class BigQueryExecutorError(Exception):
-  """Error when executor fails to run query."""
+class BigQueryExecutorError(exceptions.GarfExecutorError):
+  """Error when BigQueryExecutor fails to run query."""
 
 
 class BigQueryExecutor(query_editor.TemplateProcessorMixin):
@@ -62,26 +62,48 @@ def client(self) -> bigquery.Client:
     return bigquery.Client(self.project_id)
 
   def execute(
-    self, script_name: str, query_text: str, params: dict | None = None
-  ) -> pd.DataFrame:
+    self,
+    query: str,
+    title: str,
+    context: execution_context.ExecutionContext = (
+      execution_context.ExecutionContext()
+    ),
+  ) -> report.GarfReport:
     """Executes query in BigQuery.
 
     Args:
-        script_name: Script identifier.
-        query_text: Query to be executed.
-        params: Optional parameters to be replaced in query text.
+      query: Location of the query.
+      title: Name of the query.
+      context: Query execution context.
 
     Returns:
-        DataFrame if query returns some data otherwise empty DataFrame.
+      Report with data if query returns some data otherwise empty Report.
     """
-    query_text = self.replace_params_template(query_text, params)
+    query_text = self.replace_params_template(query, context.query_parameters)
     job = self.client.query(query_text)
     try:
       result = job.result()
-      logger.debug('%s launched successfully', script_name)
+      logger.debug('%s launched successfully', title)
       if result.total_rows:
-        return result.to_dataframe()
-      return pd.DataFrame()
+        results = report.GarfReport.from_pandas(result.to_dataframe())
+      else:
+        results = report.GarfReport()
+      if context.writer and results:
+        writer_client = context.writer_client
+        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),
+        )
+        logger.info('%s executed successfully', title)
+        return writing_result
+      return results
     except google_cloud_exceptions.GoogleCloudError as e:
       raise BigQueryExecutorError(e) from e
 
@@ -92,7 +114,7 @@ def create_datasets(self, macros: dict | None) -> None:
     are treated as dataset names.
 
     Args:
-        macros: Mapping containing data for query execution.
+      macros: Mapping containing data for query execution.
     """
     if macros and (datasets := extract_datasets(macros)):
       for dataset in datasets:

libs/garf_executors/garf_executors/entrypoints/cli.py

@@ -24,7 +24,7 @@
 from concurrent import futures
 
 import garf_executors
-from garf_executors import exceptions
+from garf_executors import bq_executor, exceptions, sql_executor
 from garf_executors.entrypoints import utils
 from garf_io import reader
 
@@ -56,13 +56,6 @@ def main():
   if args.version:
     print(garf_executors.__version__)
     sys.exit()
-  if not (source := args.source):
-    raise exceptions.GarfExecutorError(
-      f'Select one of available sources: {list(garf_executors.FETCHERS.keys())}'
-    )
-  if not (concrete_api_fetcher := garf_executors.FETCHERS.get(source)):
-    raise exceptions.GarfExecutorError(f'Source {source} is not available.')
-
   logger = utils.init_logging(
     loglevel=args.loglevel.upper(), logger_type=args.logger
   )
@@ -88,8 +81,8 @@ def main():
     writer_parameters=config.writer_params,
     fetcher_parameters=source_parameters,
   )
-  query_executor = garf_executors.api_executor.ApiQueryExecutor(
-    concrete_api_fetcher(**source_parameters)
+  query_executor = garf_executors.setup_executor(
+    args.source, context.fetcher_parameters
   )
   if args.parallel_queries:
     logger.info('Running queries in parallel')

libs/garf_executors/garf_executors/entrypoints/server.py

@@ -22,6 +22,7 @@
 
 import garf_executors
 from garf_executors import exceptions
+from garf_executors.entrypoints import utils
 from garf_io import reader
 
 
@@ -72,28 +73,17 @@ class ApiExecutorResponse(pydantic.BaseModel):
 
 @router.post('/execute')
 async def execute(request: ApiExecutorRequest) -> ApiExecutorResponse:
-  if not (concrete_api_fetcher := garf_executors.FETCHERS.get(request.source)):
-    raise exceptions.GarfExecutorError(
-      f'Source {request.source} is not available.'
-    )
-
-  query_executor = garf_executors.ApiQueryExecutor(
-    concrete_api_fetcher(**request.context.fetcher_parameters)
+  query_executor = garf_executors.setup_executor(
+    request.source, request.context.fetcher_parameters
   )
-
   result = query_executor.execute(request.query, request.title, request.context)
   return ApiExecutorResponse(results=[result])
 
 
 @router.post('/execute:batch')
 async def execute_batch(request: ApiExecutorRequest) -> ApiExecutorResponse:
-  if not (concrete_api_fetcher := garf_executors.FETCHERS.get(request.source)):
-    raise exceptions.GarfExecutorError(
-      f'Source {request.source} is not available.'
-    )
-
-  query_executor = garf_executors.ApiQueryExecutor(
-    concrete_api_fetcher(**request.context.fetcher_parameters)
+  query_executor = garf_executors.setup_executor(
+    request.source, request.context.fetcher_parameters
   )
   file_reader = reader.FileReader()
   results = []

libs/garf_executors/garf_executors/execution_context.py

@@ -0,0 +1,60 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# pylint: disable=C0330, g-bad-import-order, g-multiple-import
+
+from __future__ import annotations
+
+import pydantic
+
+from garf_core import query_editor
+from garf_io import writer
+from garf_io.writers import abs_writer
+
+
+class ExecutionContext(pydantic.BaseModel):
+  """Common context for executing one or more queries.
+
+  Attributes:
+    query_parameters: Parameters to dynamically change query text.
+    fetcher_parameters: Parameters to specify fetching setup.
+    writer: Type of writer to use.
+    writer_parameters: Optional parameters to setup writer.
+  """
+
+  query_parameters: query_editor.GarfQueryParameters | None = pydantic.Field(
+    default_factory=dict
+  )
+  fetcher_parameters: dict[str, str] | None = pydantic.Field(
+    default_factory=dict
+  )
+  writer: str | None = None
+  writer_parameters: dict[str, str] | None = pydantic.Field(
+    default_factory=dict
+  )
+
+  def model_post_init(self, __context__) -> None:
+    if self.fetcher_parameters is None:
+      self.fetcher_parameters = {}
+    if self.writer_parameters is None:
+      self.writer_parameters = {}
+
+  @property
+  def writer_client(self) -> abs_writer.AbsWriter:
+    writer_client = writer.create_writer(self.writer, **self.writer_parameters)
+    if self.writer == 'bq':
+      _ = writer_client.create_or_get_dataset()
+    if self.writer == 'sheet':
+      writer_client.init_client()
+    return writer_client