[executors] feat: add class method from_parameters to ApiQueryExecutor

* Simplify initialization of fetchers

Author: AndreyMarkinPPC

libs/garf_executors/garf_executors/__init__.py

@@ -15,17 +15,12 @@
 
 from __future__ import annotations
 
-from garf_executors import bq_executor, exceptions, sql_executor
+from garf_executors import bq_executor, fetchers, 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':
@@ -35,16 +30,16 @@ def setup_executor(source: str, fetcher_parameters: dict[str, str]):
       )
     )
   else:
+    concrete_api_fetcher = fetchers.get_report_fetcher(source)
     query_executor = ApiQueryExecutor(
       concrete_api_fetcher(**fetcher_parameters)
     )
   return query_executor
 
 
 __all__ = [
-  'FETCHERS',
   'ApiQueryExecutor',
   'ApiExecutionContext',
 ]
 
-__version__ = '0.0.7'
+__version__ = '0.0.8'

libs/garf_executors/garf_executors/api_executor.py

@@ -23,7 +23,7 @@
 import logging
 
 from garf_core import report_fetcher
-from garf_executors import exceptions, execution_context, executor
+from garf_executors import exceptions, execution_context, executor, fetchers
 
 logger = logging.getLogger(__name__)
 
@@ -49,6 +49,13 @@ def __init__(self, fetcher: report_fetcher.ApiReportFetcher) -> None:
     """
     self.fetcher = fetcher
 
+  @classmethod
+  def from_fetcher_alias(
+    cls, source: str, fetcher_parameters: dict[str, str]
+  ) -> ApiQueryExecutor:
+    concrete_api_fetcher = fetchers.get_report_fetcher(source)
+    return ApiQueryExecutor(concrete_api_fetcher(**fetcher_parameters))
+
   async def aexecute(
     self,
     query: str,

libs/garf_executors/garf_executors/fetchers.py

@@ -15,23 +15,41 @@
 import inspect
 from importlib.metadata import entry_points
 
-from garf_core import report_fetcher
+from garf_core import exceptions, report_fetcher
 
 
-def get_report_fetchers() -> dict[str, report_fetcher.ApiReportFetcher]:
+def find_fetchers() -> set[str]:
+  """Identifiers all available report fetchers."""
+  return {fetcher.name for fetcher in entry_points(group='garf')}
+
+
+def get_report_fetcher(source: str) -> type[report_fetcher.ApiReportFetcher]:
+  """Loads report fetcher for a given source.
+
+  Args:
+    source: Alias for a source associated with a fetcher.
+
+  Returns:
+    Class for a found report fetcher.
+
+  Raises:
+    ApiReportFetcherError: When fetcher cannot be loaded.
+    MissingApiReportFetcherError: When fetcher not found.
+  """
+  if source not in find_fetchers():
+    raise report_fetcher.MissingApiReportFetcherError(source)
   fetchers = entry_points(group='garf')
-  found_fetchers = {}
   for fetcher in fetchers:
-    try:
-      fetcher_module = fetcher.load()
-      for name, obj in inspect.getmembers(fetcher_module):
-        if inspect.isclass(obj) and issubclass(
-          obj, report_fetcher.ApiReportFetcher
-        ):
-          found_fetchers[fetcher.name] = getattr(fetcher_module, name)
-    except ModuleNotFoundError:
-      continue
-  return found_fetchers
-
-
-FETCHERS = get_report_fetchers()
+    if fetcher.name == source:
+      try:
+        fetcher_module = fetcher.load()
+        for name, obj in inspect.getmembers(fetcher_module):
+          if inspect.isclass(obj) and issubclass(
+            obj, report_fetcher.ApiReportFetcher
+          ):
+            return getattr(fetcher_module, name)
+      except ModuleNotFoundError:
+        continue
+  raise exceptions.ApiReportFetcherError(
+    f'No fetcher available for the source "{source}"'
+  )