[Core] Add a util to cache results from coroutines

CHANGELOG.md

@@ -7,13 +7,22 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 <!-- towncrier release notes start -->
 
+## 0.14.1 (2024-11-13)
+
+
+### Improvements
+
+- Added a decorator to help with caching results from coroutines.
+
+
 ## 0.14.0 (2024-11-12)
 
 
 ### Improvements
 
 - Add support for choosing default resources that the integration will create dynamically
 
+
 ## 0.13.1 (2024-11-12)
 
 
@@ -29,6 +38,7 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 - Bump python from 3.11 to 3.12 (0.13.0)
 
+
 ## 0.12.9 (2024-11-07)
 
 

port_ocean/tests/utils/test_cache.py

@@ -0,0 +1,189 @@
+from typing import Any
+import asyncio
+from port_ocean.utils import cache  # Import the module where 'event' is used
+import pytest
+from dataclasses import dataclass, field
+from typing import AsyncGenerator, AsyncIterator, List, TypeVar
+
+
+@dataclass
+class EventContext:
+    attributes: dict[str, Any] = field(default_factory=dict)
+
+
+@pytest.fixture
+def event() -> EventContext:
+    return EventContext()
+
+
+T = TypeVar("T")
+
+
+async def collect_iterator_results(iterator: AsyncIterator[List[T]]) -> List[T]:
+    results = []
+    async for item in iterator:
+        results.extend(item)
+    return results
+
+
+@pytest.mark.asyncio
+async def test_cache_iterator_result(event: EventContext, monkeypatch: Any) -> None:
+    monkeypatch.setattr(cache, "event", event)
+
+    call_count = 0
+
+    @cache.cache_iterator_result()
+    async def sample_iterator(x: int) -> AsyncGenerator[List[int], None]:
+        nonlocal call_count
+        call_count += 1
+        for i in range(x):
+            await asyncio.sleep(0.1)
+            yield [i]
+
+    result1 = await collect_iterator_results(sample_iterator(3))
+    assert result1 == [0, 1, 2]
+    assert call_count == 1
+
+    result2 = await collect_iterator_results(sample_iterator(3))
+    assert result2 == [0, 1, 2]
+    assert call_count == 1
+
+    result3 = await collect_iterator_results(sample_iterator(4))
+    assert result3 == [0, 1, 2, 3]
+    assert call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_cache_iterator_result_with_kwargs(
+    event: EventContext, monkeypatch: Any
+) -> None:
+    monkeypatch.setattr(cache, "event", event)
+
+    call_count = 0
+
+    @cache.cache_iterator_result()
+    async def sample_iterator(x: int, y: int = 1) -> AsyncGenerator[List[int], None]:
+        nonlocal call_count
+        call_count += 1
+        for i in range(x * y):
+            await asyncio.sleep(0.1)
+            yield [i]
+
+    result1 = await collect_iterator_results(sample_iterator(2, y=2))
+    assert result1 == [0, 1, 2, 3]
+    assert call_count == 1
+
+    result2 = await collect_iterator_results(sample_iterator(2, y=2))
+    assert result2 == [0, 1, 2, 3]
+    assert call_count == 1
+
+    result3 = await collect_iterator_results(sample_iterator(2, y=3))
+    assert result3 == [0, 1, 2, 3, 4, 5]
+    assert call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_cache_iterator_result_no_cache(
+    event: EventContext, monkeypatch: Any
+) -> None:
+    monkeypatch.setattr(cache, "event", event)
+
+    call_count = 0
+
+    @cache.cache_iterator_result()
+    async def sample_iterator(x: int) -> AsyncGenerator[List[int], None]:
+        nonlocal call_count
+        call_count += 1
+        for i in range(x):
+            await asyncio.sleep(0.1)
+            yield [i]
+
+    result1 = await collect_iterator_results(sample_iterator(3))
+    assert result1 == [0, 1, 2]
+    assert call_count == 1
+
+    event.attributes.clear()
+
+    result2 = await collect_iterator_results(sample_iterator(3))
+    assert result2 == [0, 1, 2]
+    assert call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_cache_coroutine_result(event: EventContext, monkeypatch: Any) -> None:
+    monkeypatch.setattr(cache, "event", event)
+
+    call_count = 0
+
+    @cache.cache_coroutine_result()
+    async def sample_coroutine(x: int) -> int:
+        nonlocal call_count
+        call_count += 1
+        await asyncio.sleep(0.1)
+        return x * 2
+
+    result1 = await sample_coroutine(2)
+    assert result1 == 4
+    assert call_count == 1
+
+    result2 = await sample_coroutine(2)
+    assert result2 == 4
+    assert call_count == 1
+
+    result3 = await sample_coroutine(3)
+    assert result3 == 6
+    assert call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_cache_coroutine_result_with_kwargs(
+    event: EventContext, monkeypatch: Any
+) -> None:
+    monkeypatch.setattr(cache, "event", event)
+
+    call_count = 0
+
+    @cache.cache_coroutine_result()
+    async def sample_coroutine(x: int, y: int = 1) -> int:
+        nonlocal call_count
+        call_count += 1
+        await asyncio.sleep(0.1)
+        return x * y
+
+    result1 = await sample_coroutine(2, y=3)
+    assert result1 == 6
+    assert call_count == 1
+
+    result2 = await sample_coroutine(2, y=3)
+    assert result2 == 6
+    assert call_count == 1
+
+    result3 = await sample_coroutine(2, y=4)
+    assert result3 == 8
+    assert call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_cache_coroutine_result_no_cache(
+    event: EventContext, monkeypatch: Any
+) -> None:
+    monkeypatch.setattr(cache, "event", event)
+
+    call_count = 0
+
+    @cache.cache_coroutine_result()
+    async def sample_coroutine(x: int) -> int:
+        nonlocal call_count
+        call_count += 1
+        await asyncio.sleep(0.1)
+        return x * 2
+
+    result1 = await sample_coroutine(2)
+    assert result1 == 4
+    assert call_count == 1
+
+    event.attributes.clear()
+
+    result2 = await sample_coroutine(2)
+    assert result2 == 4
+    assert call_count == 2

port_ocean/utils/cache.py

@@ -1,9 +1,10 @@
 import functools
 import hashlib
-from typing import Callable, AsyncIterator, Any
+from typing import Callable, AsyncIterator, Awaitable, Any
 from port_ocean.context.event import event
 
 AsyncIteratorCallable = Callable[..., AsyncIterator[list[Any]]]
+AsyncCallable = Callable[..., Awaitable[Any]]
 
 
 def hash_func(function_name: str, *args: Any, **kwargs: Any) -> str:
@@ -59,3 +60,38 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
         return wrapper
 
     return decorator
+
+
+def cache_coroutine_result() -> Callable[[AsyncCallable], AsyncCallable]:
+    """Coroutine version of `cache_iterator_result` from port_ocean.utils.cache
+
+    Decorator that caches the result of a coroutine function.
+    It checks if the result is already in the cache, and if not,
+    fetches the result, caches it, and returns the cached value.
+
+    The cache is stored in the scope of the running event and is
+    removed when the event is finished.
+
+    Usage:
+    ```python
+    @cache_coroutine_result()
+    async def my_coroutine_function():
+        # Your code here
+    ```
+    """
+
+    def decorator(func: AsyncCallable) -> AsyncCallable:
+        @functools.wraps(func)
+        async def wrapper(*args: Any, **kwargs: Any) -> Any:
+            cache_key = hash_func(func.__name__, *args, **kwargs)
+
+            if cache := event.attributes.get(cache_key):
+                return cache
+
+            result = await func(*args, **kwargs)
+            event.attributes[cache_key] = result
+            return result
+
+        return wrapper
+
+    return decorator

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "port-ocean"
-version = "0.14.0"
+version = "0.14.1"
 description = "Port Ocean is a CLI tool for managing your Port projects."
 readme = "README.md"
 homepage = "https://app.getport.io"