fix: Disable caching of task status endpoint

idf_component_tools/registry/api_client.py

@@ -73,6 +73,7 @@ def request(
                 headers: t.Optional[t.Dict] = None,
                 schema: t.Optional[ApiBaseModel] = None,
                 timeout: t.Optional[t.Union[float, t.Tuple[float, float]]] = None,
+                do_not_cache=False,
             ):
                 # always access '<registry_url>/api' while doing api calls
                 path = ['api', *path]
@@ -87,6 +88,7 @@ def request(
                     headers=headers,
                     schema=schema,
                     timeout=timeout,
+                    do_not_cache=do_not_cache,
                 )
 
             return f(self, request=request, *args, **kwargs)
@@ -186,4 +188,10 @@ def yank_version(
 
     @_request
     def task_status(self, request: t.Callable, job_id: str) -> TaskStatus:
-        return TaskStatus.model_validate(request('get', ['tasks', job_id]))
+        return TaskStatus.model_validate(
+            request(
+                'get',
+                ['tasks', job_id],
+                do_not_cache=True,
+            )
+        )

idf_component_tools/registry/request_processor.py

@@ -63,10 +63,15 @@ def cache_request(func):
     """Decorator to conditionally cache GET and HEAD requests based on CACHE_HTTP_REQUESTS"""
 
     def wrapper(*args, **kwargs):
-        if ComponentManagerSettings().CACHE_HTTP_REQUESTS and kwargs.get('method', '').lower() in [
-            'get',
-            'head',
-        ]:
+        do_not_cache = kwargs.pop('do_not_cache', False)
+
+        cache_conditions = [
+            ComponentManagerSettings().CACHE_HTTP_REQUESTS,
+            kwargs.get('method', '').lower() in ['get', 'head'],
+            do_not_cache is False,
+        ]
+
+        if all(cache_conditions):
             return cache_to_dict(_request_cache, func, *args, **kwargs)
         return func(*args, **kwargs)
 
@@ -182,6 +187,7 @@ def base_request(
     schema: t.Optional[ApiBaseModel] = None,
     timeout: t.Optional[t.Union[float, t.Tuple[float, float]]] = None,
     use_storage: bool = False,
+    do_not_cache=False,
 ) -> t.Dict:
     endpoint = join_url(url, *path)
 
@@ -192,7 +198,16 @@ def base_request(
     if request_timeout is None:
         request_timeout = DEFAULT_REQUEST_TIMEOUT
 
-    response = make_request(session, endpoint, data, json, headers, request_timeout, method=method)
+    response = make_request(
+        session,
+        endpoint,
+        data,
+        json,
+        headers,
+        request_timeout,
+        method=method,
+        do_not_cache=do_not_cache,
+    )
     response_json = handle_response_errors(response, endpoint, use_storage)
 
     if schema is None:

tests/idf_comonent_tools_tests/registry/test_request_processor.py

@@ -126,3 +126,22 @@ def test_cache_request_caches_only_get_and_head(mocker):
     ]
     assert mock_func.call_args_list == expected_calls
     assert len(_request_cache) == 2  # Cache should have entries for GET and HEAD
+
+
+@pytest.mark.enable_request_cache
+def test_cache_request_with_do_not_cache(mocker):
+    # Mock function to be decorated
+    mock_func = mocker.Mock(return_value='response')
+    decorated_func = cache_request(mock_func)
+
+    # Clear the cache before testing
+    _request_cache.clear()
+
+    # Call the function with do_not_cache set to True
+    result1 = decorated_func(method='GET', url='http://example.com', do_not_cache=True)
+    result2 = decorated_func(method='GET', url='http://example.com', do_not_cache=True)
+
+    assert result1 == 'response'
+    assert result2 == 'response'
+    assert mock_func.call_count == 2  # Should be called twice since caching is disabled
+    assert len(_request_cache) == 0  # Cache should remain empty