[core] feat: add support for caching fetching results

AndreyMarkinPPC · AndreyMarkinPPC · commit dd74a80c21a7 · 2025-11-19T13:22:30.000+04:00
diff --git a/libs/core/garf_core/cache.py b/libs/core/garf_core/cache.py
@@ -0,0 +1,115 @@
+# 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.
+
+"""Stores and loads reports from a cache instead of calling API."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import pathlib
+from typing import Final
+
+from garf_core import exceptions, query_editor, report
+
+logger = logging.getLogger(__name__)
+
+
+class GarfCacheFileNotFoundError(exceptions.GarfError):
+  """Exception for not found cached report."""
+
+
+DEFAULT_CACHE_LOCATION: Final[str] = os.getenv(
+  'GARF_CACHE_LOCATION', str(pathlib.Path.home() / '.garf/cache/')
+)
+
+
+class GarfCache:
+  """Stores and loads reports from a cache instead of calling API.
+
+  Attribute:
+    location: Folder where cached results are stored.
+  """
+
+  def __init__(
+    self,
+    location: str | None = None,
+  ) -> None:
+    """Stores and loads reports from a cache instead of calling API.
+
+    Args:
+      location: Folder where cached results are stored.
+    """
+    self.location = pathlib.Path(location or DEFAULT_CACHE_LOCATION)
+
+  def load(
+    self, query: query_editor.BaseQueryElements, args=None, kwargs=None
+  ) -> report.GarfReport:
+    """Loads report from cache based on query definition.
+
+    Args:
+      query: Query elements.
+      args: Query parameters.
+      kwargs: Optional keyword arguments.
+
+    Returns:
+      Cached report.
+
+    Raises:
+      GarfCacheFileNotFoundError: If cached report not found
+    """
+    args_hash = args.hash if args else ''
+    kwargs_hash = (
+      hashlib.md5(json.dumps(kwargs).encode('utf-8')).hexdigest()
+      if kwargs
+      else ''
+    )
+    hash_identifier = f'{query.hash}:{args_hash}:{kwargs_hash}'
+    cached_path = self.location / f'{hash_identifier}.json'
+    if cached_path.exists():
+      with open(cached_path, 'r', encoding='utf-8') as f:
+        data = json.load(f)
+      logger.debug('Report is loaded from cache: %s', str(cached_path))
+      return report.GarfReport.from_json(data)
+    raise GarfCacheFileNotFoundError
+
+  def save(
+    self,
+    fetched_report: report.GarfReport,
+    query: query_editor.BaseQueryElements,
+    args=None,
+    kwargs=None,
+  ) -> None:
+    """Saves report to cache based on query definition.
+
+    Args:
+      fetched_report: Report to save.
+      query: Query elements.
+      args: Query parameters.
+      kwargs: Optional keyword arguments.
+    """
+    self.location.mkdir(parents=True, exist_ok=True)
+    args_hash = args.hash if args else ''
+    kwargs_hash = (
+      hashlib.md5(json.dumps(kwargs).encode('utf-8')).hexdigest()
+      if kwargs
+      else ''
+    )
+    hash_identifier = f'{query.hash}:{args_hash}:{kwargs_hash}'
+    cached_path = self.location / f'{hash_identifier}.json'
+    logger.debug('Report is saved to cache: %s', str(cached_path))
+    with open(cached_path, 'w', encoding='utf-8') as f:
+      json.dump(fetched_report.to_json(), f)
diff --git a/libs/core/garf_core/query_editor.py b/libs/core/garf_core/query_editor.py
@@ -15,8 +15,9 @@
 
 from __future__ import annotations
 
-import dataclasses
 import datetime
+import hashlib
+import json
 import logging
 import re
 from typing import Generator, Union
@@ -39,6 +40,11 @@ class GarfQueryParameters(pydantic.BaseModel):
   macro: QueryParameters = pydantic.Field(default_factory=dict)
   template: QueryParameters = pydantic.Field(default_factory=dict)
 
+  @property
+  def hash(self) -> str:
+    hash_fields = self.model_dump(exclude_none=True)
+    return hashlib.md5(json.dumps(hash_fields).encode('utf-8')).hexdigest()
+
 
 class GarfMacroError(query_parser.GarfQueryError):
   """Specifies incorrect macro in Garf query."""
@@ -52,33 +58,32 @@ class GarfBuiltInQueryError(query_parser.GarfQueryError):
   """Specifies non-existing builtin query."""
 
 
-@dataclasses.dataclass
-class BaseQueryElements:
+class BaseQueryElements(pydantic.BaseModel):
   """Contains raw query and parsed elements.
 
   Attributes:
-      title: Title of the query that needs to be parsed.
-      text: Text of the query that needs to be parsed.
-      resource_name: Name of Google Ads API reporting resource.
-      fields: Ads API fields that need to be fetched.
-      column_names: Friendly names for fields which are used when saving data
-      column_names: Friendly names for fields which are used when saving data
-      customizers: Attributes of fields that need to be be extracted.
-      virtual_columns: Attributes of fields that need to be be calculated.
-      is_builtin_query: Whether query is built-in.
+    title: Title of the query that needs to be parsed.
+    text: Text of the query that needs to be parsed.
+    resource_name: Name of Google Ads API reporting resource.
+    fields: Ads API fields that need to be fetched.
+    column_names: Friendly names for fields which are used when saving data
+    column_names: Friendly names for fields which are used when saving data
+    customizers: Attributes of fields that need to be be extracted.
+    virtual_columns: Attributes of fields that need to be be calculated.
+    is_builtin_query: Whether query is built-in.
   """
 
-  title: str
+  title: str | None
   text: str
   resource_name: str | None = None
-  fields: list[str] = dataclasses.field(default_factory=list)
-  filters: list[str] = dataclasses.field(default_factory=list)
-  sorts: list[str] = dataclasses.field(default_factory=list)
-  column_names: list[str] = dataclasses.field(default_factory=list)
-  customizers: dict[str, dict[str, str]] = dataclasses.field(
+  fields: list[str] = pydantic.Field(default_factory=list)
+  filters: list[str] = pydantic.Field(default_factory=list)
+  sorts: list[str] = pydantic.Field(default_factory=list)
+  column_names: list[str] = pydantic.Field(default_factory=list)
+  customizers: dict[str, query_parser.Customizer] = pydantic.Field(
     default_factory=dict
   )
-  virtual_columns: dict[str, query_parser.VirtualColumn] = dataclasses.field(
+  virtual_columns: dict[str, query_parser.VirtualColumn] = pydantic.Field(
     default_factory=dict
   )
   is_builtin_query: bool = False
@@ -87,12 +92,16 @@ def __eq__(self, other: BaseQueryElements) -> bool:  # noqa: D105
     return (
       self.column_names,
       self.fields,
+      self.filters,
+      self.sorts,
       self.resource_name,
       self.customizers,
       self.virtual_columns,
     ) == (
       other.column_names,
       other.fields,
+      other.filters,
+      other.sorts,
       other.resource_name,
       other.customizers,
       other.virtual_columns,
@@ -103,6 +112,11 @@ def request(self) -> str:
     """API request."""
     return ','.join(self.fields)
 
+  @property
+  def hash(self) -> str:
+    hash_fields = self.model_dump(exclude_none=True, exclude={'title', 'text'})
+    return hashlib.md5(json.dumps(hash_fields).encode('utf-8')).hexdigest()
+
 
 class CommonParametersMixin:
   """Helper mixin to inject set of common parameters to all queries."""
diff --git a/libs/core/garf_core/report_fetcher.py b/libs/core/garf_core/report_fetcher.py
@@ -23,12 +23,14 @@
 
 import asyncio
 import logging
+import pathlib
 from typing import Callable
 
 from opentelemetry import trace
 
 from garf_core import (
   api_clients,
+  cache,
   exceptions,
   parsers,
   query_editor,
@@ -63,6 +65,7 @@ class ApiReportFetcher:
       query_specification_builder: Class to perform query parsing.
       builtin_queries:
         Mapping between query name and function for generating GarfReport.
+      enable_cache: Whether to load / save report from / to cache.
   """
 
   def __init__(
@@ -74,6 +77,8 @@ def __init__(
     ),
     builtin_queries: dict[str, Callable[[ApiReportFetcher], report.GarfReport]]
     | None = None,
+    enable_cache: bool = False,
+    cache_path: str | pathlib.Path | None = None,
     **kwargs: str,
   ) -> None:
     """Instantiates ApiReportFetcher based on provided api client.
@@ -84,11 +89,15 @@ def __init__(
       query_specification_builder: Class to perform query parsing.
       builtin_queries:
         Mapping between query name and function for generating GarfReport.
+      enable_cache: Whether to load / save report from / to cache.
+      cache_path: Optional path to cache folder.
     """
     self.api_client = api_client
     self.parser = parser
     self.query_specification_builder = query_specification_builder
     self.query_args = kwargs
+    self.enable_cache = enable_cache
+    self.cache = cache.GarfCache(cache_path)
     self.builtin_queries = builtin_queries or {}
 
   def add_builtin_queries(
@@ -156,13 +165,24 @@ def fetch(
         )
       return builtin_report(self, **kwargs)
 
+    if self.enable_cache:
+      try:
+        cached_report = self.cache.load(query, args, kwargs)
+        logger.warning('Cached version of query is loaded')
+        span.set_attribute('is_cached_query', True)
+        return cached_report
+      except cache.GarfCacheFileNotFoundError:
+        logger.debug('Cached version not found, generating')
     response = self.api_client.call_api(query, **kwargs)
     if not response:
       return report.GarfReport(query_specification=query)
 
     parsed_response = self.parser(query).parse_response(response)
-    return report.GarfReport(
+    fetched_report = report.GarfReport(
       results=parsed_response,
       column_names=query.column_names,
       query_specification=query,
     )
+    if self.enable_cache:
+      self.cache.save(fetched_report, query, args, kwargs)
+    return fetched_report
diff --git a/libs/core/tests/unit/test_cache.py b/libs/core/tests/unit/test_cache.py
@@ -0,0 +1,33 @@
+# 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.
+
+import pytest
+from garf_core import cache, query_editor, report
+
+
+class TestGarfCache:
+  @pytest.fixture()
+  def cache(self, tmp_path):
+    return cache.GarfCache(str(tmp_path))
+
+  def test_save(self, cache):
+    test_report = report.GarfReport(results=[[1]], column_names=['test'])
+    query = query_editor.QuerySpecification(
+      text='SELECT test FROM test'
+    ).generate()
+
+    cache.save(test_report, query)
+    loaded_report = cache.load(query)
+
+    assert loaded_report == test_report
diff --git a/libs/core/tests/unit/test_report_fetcher.py b/libs/core/tests/unit/test_report_fetcher.py
@@ -14,6 +14,7 @@
 
 
 import datetime
+import logging
 
 import pytest
 from garf_core import (
@@ -51,6 +52,38 @@ def test_fetch_returns_correct_report_for_dict_parser(
 
     assert test_report == expected_report
 
+  def test_fetch_returns_saves_and_loads_cached_version(self, caplog, tmp_path):
+    test_api_client = api_clients.FakeApiClient(
+      results=[
+        {'column': {'name': 1}, 'other_column': 2},
+        {'column': {'name': 2}, 'other_column': 2},
+        {'column': {'name': 3}, 'other_column': 2},
+      ]
+    )
+    test_fetcher = report_fetcher.ApiReportFetcher(
+      api_client=test_api_client,
+      parser=parsers.DictParser,
+      enable_cache=True,
+      cache_path=tmp_path,
+    )
+    query = 'SELECT column.name, other_column FROM test'
+    expected_report = report.GarfReport(
+      results=[[1, 2], [2, 2], [3, 2]],
+      column_names=['column_name', 'other_column'],
+    )
+
+    with caplog.at_level(logging.DEBUG):
+      test_report = test_fetcher.fetch(query)
+      assert 'Report is saved to cache' in caplog.text
+      assert 'Cached version not found, generating' in caplog.text
+      assert test_report == expected_report
+
+    with caplog.at_level(logging.DEBUG):
+      test_report = test_fetcher.fetch(query)
+      assert 'Report is loaded from cache' in caplog.text
+      assert 'Cached version of query is loaded' in caplog.text
+      assert test_report == expected_report
+
   def test_fetch_returns_empty_report_for_empty_api_response(self):
     test_api_client = api_clients.FakeApiClient(results=[])
     fetcher = report_fetcher.ApiReportFetcher(
diff --git a/libs/executors/garf_executors/__init__.py b/libs/executors/garf_executors/__init__.py
@@ -24,7 +24,9 @@
 
 @tracer.start_as_current_span('setup_executor')
 def setup_executor(
-  source: str, fetcher_parameters: dict[str, str]
+  source: str,
+  fetcher_parameters: dict[str, str | int | bool],
+  enable_cache: bool = False,
 ) -> type[executor.Executor]:
   """Initializes executors based on a source and parameters."""
   if source == 'bq':
@@ -40,7 +42,7 @@ def setup_executor(
   else:
     concrete_api_fetcher = fetchers.get_report_fetcher(source)
     query_executor = ApiQueryExecutor(
-      concrete_api_fetcher(**fetcher_parameters)
+      concrete_api_fetcher(**fetcher_parameters, enable_cache=enable_cache)
     )
   return query_executor
 
diff --git a/libs/executors/garf_executors/entrypoints/cli.py b/libs/executors/garf_executors/entrypoints/cli.py
diff --git a/libs/executors/garf_executors/execution_context.py b/libs/executors/garf_executors/execution_context.py
