fix: read local Actor input from the storage directory with non-filesystem storage clients

docs/02_concepts/02_actor_input.mdx

@@ -22,7 +22,7 @@ For example, if an Actor received a JSON input with two fields, `{ "firstNumber"
 
 ## Input file in the working directory
 
-When you run the Actor locally and its default key-value store holds no input record, `Actor.get_input` reads the input from an `INPUT.json` or a bare `INPUT` file in the current working directory. The file name follows the configured input key. The `.json` file is parsed as JSON. The bare file is parsed as JSON when it's valid JSON and returned as `bytes` otherwise. If both files exist, `Actor.get_input` raises an error. On the Apify platform, the working directory is never read.
+When you run the Actor locally and its default key-value store holds no input record, `Actor.get_input` reads the input from an `INPUT.json` or a bare `INPUT` file in the current working directory, and then in the default key-value store directory (`<storage dir>/key_value_stores/<default store ID>`, `storage/key_value_stores/default` by default, where `apify run` writes it). The latter covers storage clients that don't read that directory, such as in-memory storage. The working directory takes precedence. The file name follows the configured input key. The `.json` file is parsed as JSON. The bare file is parsed as JSON when it's valid JSON and returned as `bytes` otherwise. If both files exist in the same directory, `Actor.get_input` raises an error. On the Apify platform, neither directory is read.
 
 ## Validating input
 

src/apify/_actor.py

@@ -785,21 +785,35 @@ async def get_input(self) -> Any:
         plaintext before being returned.
 
         When running locally and the default key-value store holds no input record, the input is read from a
-        `<input_key>` or `<input_key>.json` file in the current working directory instead. The `.json` file is parsed
-        as JSON; the bare file is parsed as JSON when it is valid JSON, and returned as `bytes` otherwise.
+        `<input_key>` or `<input_key>.json` file, looked up first in the current working directory and then in the
+        default key-value store's directory under the storage directory, where `apify run` writes it. The latter only
+        matters for storage clients that don't read that directory themselves. The `.json` file is parsed as JSON; the
+        bare file is parsed as JSON when it is valid JSON, and returned as `bytes` otherwise.
 
         Returns:
             The Actor input, usually a `dict` of input fields, or `None` if the Actor has no input.
 
         Raises:
-            RuntimeError: If both `<input_key>` and `<input_key>.json` exist in the working directory.
-            ValueError: If the `<input_key>.json` file in the working directory is not valid JSON.
+            RuntimeError: If both `<input_key>` and `<input_key>.json` exist in the same directory.
+            ValueError: If the `<input_key>.json` file is not valid JSON.
         """
         input_key = self.configuration.input_key
         key_value_store = await self.open_key_value_store()
         input_value = await key_value_store.get_value(input_key)
         if input_value is None and not self.is_at_home() and not await key_value_store.record_exists(input_key):
-            input_value = await self._read_input_from_working_directory(input_key)
+            input_path = await self._find_input_file(input_key, Path.cwd(), 'working directory')
+            if input_path is None:
+                # Mirrors apify-cli's `getLocalKeyValueStorePath`, which is where `apify run` writes the input.
+                default_kvs_dir = (
+                    Path(self.configuration.storage_dir)
+                    / 'key_value_stores'
+                    / (self.configuration.default_key_value_store_id or 'default')
+                )
+                input_path = await self._find_input_file(
+                    input_key, default_kvs_dir, 'default key-value store directory'
+                )
+            if input_path is not None:
+                input_value = await self._read_input_file(input_path, input_key)
 
         input_secrets_private_key = self.configuration.input_secrets_private_key_file
         input_secrets_key_passphrase = self.configuration.input_secrets_private_key_passphrase
@@ -1850,19 +1864,20 @@ def _get_default_exit_process(self) -> bool:
         return True
 
     @staticmethod
-    async def _read_input_from_working_directory(input_key: str) -> Any:
-        """Read the Actor input from a `<input_key>` or `<input_key>.json` file in the current working directory."""
-        candidates = [Path.cwd() / input_key, Path.cwd() / f'{input_key}.json']
+    async def _find_input_file(input_key: str, directory: Path, label: str) -> Path | None:
+        """Find the `<input_key>` or `<input_key>.json` input file in the given directory, if there is one."""
+        candidates = [directory / input_key, directory / f'{input_key}.json']
         found = [path for path in candidates if await asyncio.to_thread(path.is_file)]
 
         if len(found) > 1:
             names = ', '.join(f'"{path.name}"' for path in found)
-            raise RuntimeError(f'Found multiple input files in the working directory: {names}. Keep only one of them.')
+            raise RuntimeError(f'Found multiple input files in the {label}: {names}. Keep only one of them.')
 
-        if not found:
-            return None
+        return found[0] if found else None
 
-        input_path = found[0]
+    @staticmethod
+    async def _read_input_file(input_path: Path, input_key: str) -> Any:
+        """Read the Actor input from the given `<input_key>` or `<input_key>.json` input file."""
         content = await asyncio.to_thread(input_path.read_bytes)
 
         try:

tests/unit/actor/test_actor_key_value_store.py

@@ -1,18 +1,18 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from pathlib import Path
 
 import pytest
 
+from crawlee import service_locator
 from crawlee._utils.file import json_dumps
+from crawlee.storage_clients import MemoryStorageClient
 
 from ..._utils import PRIVATE_KEY_PASSWORD, PRIVATE_KEY_PEM_BASE64, PUBLIC_KEY, poll_until_condition
-from apify import Actor
+from apify import Actor, Configuration
 from apify._consts import ENCRYPTED_JSON_VALUE_PREFIX, ENCRYPTED_STRING_VALUE_PREFIX, ActorEnvVars, ApifyEnvVars
 from apify._crypto import public_encrypt
-
-if TYPE_CHECKING:
-    from pathlib import Path
+from apify.storage_clients import SmartApifyStorageClient
 
 
 async def test_open_returns_same_references() -> None:
@@ -268,14 +268,85 @@ async def test_get_input_raises_on_multiple_working_directory_files(
     (tmp_path / 'INPUT.json').write_text('{}')
 
     async with Actor as actor:
-        with pytest.raises(RuntimeError, match='multiple input files'):
+        with pytest.raises(RuntimeError, match='multiple input files in the working directory'):
             await actor.get_input()
 
 
-async def test_get_input_skips_working_directory_on_platform(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
-    """On the Apify platform, the working directory is never read."""
+def _use_memory_storage(store_id: str = 'default') -> Path:
+    """Switch to in-memory storage, which never reads the storage directory, and create the default store dir."""
+    service_locator.set_storage_client(SmartApifyStorageClient(local_storage_client=MemoryStorageClient()))
+    default_kvs_dir = Path(Configuration.get_global_configuration().storage_dir) / 'key_value_stores' / store_id
+    default_kvs_dir.mkdir(parents=True)
+    return default_kvs_dir
+
+
+async def test_get_input_falls_back_to_default_kvs_directory(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """Without an input record, `INPUT.json` in the default store directory is read after the working directory."""
     monkeypatch.chdir(tmp_path)
-    (tmp_path / 'INPUT.json').write_text('{"from": "file"}')
+    (_use_memory_storage() / 'INPUT.json').write_text('{"from": "storage"}')
+
+    async with Actor as actor:
+        assert await actor.get_input() == {'from': 'storage'}
+
+        (tmp_path / 'INPUT.json').write_text('{"from": "cwd"}')
+        assert await actor.get_input() == {'from': 'cwd'}
+
+
+async def test_get_input_skips_storage_directory_root(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """An input file in the root of the storage directory is not read."""
+    working_dir = tmp_path / 'working_dir'
+    working_dir.mkdir()
+    monkeypatch.chdir(working_dir)
+    _use_memory_storage()
+    (tmp_path / 'INPUT.json').write_text('{"from": "storage root"}')
+
+    async with Actor as actor:
+        assert await actor.get_input() is None
+
+
+async def test_get_input_default_kvs_directory_follows_store_id(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """The default store directory is named after the configured default key-value store ID."""
+    monkeypatch.setenv(ActorEnvVars.DEFAULT_KEY_VALUE_STORE_ID, 'custom')
+    monkeypatch.chdir(tmp_path)
+    (_use_memory_storage('custom') / 'INPUT.json').write_text('{"from": "custom store"}')
+
+    async with Actor as actor:
+        assert await actor.get_input() == {'from': 'custom store'}
+
+
+async def test_get_input_null_working_directory_file_skips_default_kvs_directory(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A working-directory file holding JSON `null` counts as present, so the default store directory is not read."""
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / 'INPUT.json').write_text('null')
+    (_use_memory_storage() / 'INPUT.json').write_text('{"from": "storage"}')
+
+    async with Actor as actor:
+        assert await actor.get_input() is None
+
+
+async def test_get_input_raises_on_multiple_default_kvs_directory_files(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """Both `INPUT` and `INPUT.json` in the default store directory raise an error naming that directory."""
+    monkeypatch.chdir(tmp_path)
+    default_kvs_dir = _use_memory_storage()
+    (default_kvs_dir / 'INPUT').write_text('{}')
+    (default_kvs_dir / 'INPUT.json').write_text('{}')
+
+    async with Actor as actor:
+        with pytest.raises(RuntimeError, match='multiple input files in the default key-value store directory'):
+            await actor.get_input()
+
+
+async def test_get_input_skips_input_files_on_platform(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """On the Apify platform, neither the working directory nor the default store directory is read."""
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / 'INPUT.json').write_text('{"from": "cwd"}')
+    (_use_memory_storage() / 'INPUT.json').write_text('{"from": "storage"}')
 
     async with Actor as actor:
         monkeypatch.setattr(actor, 'is_at_home', lambda: True)