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, `Actor.get_input` also accepts 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. With a storage client that doesn't read the default key-value store directory itself, such as in-memory storage, the same file is also picked up from `<storage dir>/key_value_stores/<default store ID>` (`storage/key_value_stores/default` by default), where `apify run` writes it. The input must be in exactly one place: both files in one directory, or a file in the working directory next to an input in the default key-value store, make `Actor.get_input` raise an error instead of picking one. On the Apify platform, the working directory is never read.
 
 ## Validating input
 

src/apify/_actor.py

@@ -784,22 +784,49 @@ async def get_input(self) -> Any:
         a `dict` keyed by the fields declared in the Actor's input schema. Any secret input fields are decrypted to
         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.
+        When running locally, the input may instead be a `<input_key>` or `<input_key>.json` file in the current
+        working directory. A storage client that doesn't read the default key-value store's directory under the
+        storage directory (e.g. in-memory storage) also gets the file `apify run` writes there. The input must be in
+        exactly one place. 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 the input is found in more than one place: both `<input_key>` and `<input_key>.json` in
+                one directory, or in the default key-value store and in the working 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)
+        if not self.is_at_home():
+            store_has_input = input_value is not None or await key_value_store.record_exists(input_key)
+            input_path = None
+            # Under the file-system storage client the record *is* the store directory's file, so don't read it twice.
+            if not store_has_input:
+                # 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'
+                )
+
+            cwd_input_path = await self._find_input_file(input_key, Path.cwd(), 'working directory')
+            if cwd_input_path is not None:
+                if store_has_input or input_path is not None:
+                    raise RuntimeError(
+                        f'Found the input both in the default key-value store and in a "{input_key}" or '
+                        f'"{input_key}.json" file in the working directory. Keep only one of them.'
+                    )
+                input_path = cwd_input_path
+
+            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 +1877,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:
@@ -235,28 +235,30 @@ async def test_get_input_working_directory_file_follows_input_key(
         assert await actor.get_input() == {'from': 'cli'}
 
 
-async def test_get_input_prefers_store_record_over_working_directory(
+async def test_get_input_raises_on_store_record_and_working_directory_file(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
-    """An input record in the default key-value store wins over a file in the working directory."""
+    """An input record in the default key-value store next to a file in the working directory raises an error."""
     monkeypatch.chdir(tmp_path)
     (tmp_path / 'INPUT.json').write_text('{"from": "file"}')
 
     async with Actor as actor:
         await actor.set_value('INPUT', {'from': 'store'})
-        assert await actor.get_input() == {'from': 'store'}
+        with pytest.raises(RuntimeError, match='both in the default key-value store and in'):
+            await actor.get_input()
 
 
-async def test_get_input_null_store_record_skips_working_directory(
+async def test_get_input_null_store_record_conflicts_with_working_directory_file(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
-    """An input record holding JSON `null` counts as present, so the working-directory file is not read."""
+    """An input record holding JSON `null` counts as present, so a working-directory file next to it raises."""
     monkeypatch.chdir(tmp_path)
     (tmp_path / 'INPUT.json').write_text('{"from": "file"}')
 
     async with Actor as actor:
         await actor.set_value('INPUT', None)
-        assert await actor.get_input() is None
+        with pytest.raises(RuntimeError, match='both in the default key-value store and in'):
+            await actor.get_input()
 
 
 async def test_get_input_raises_on_multiple_working_directory_files(
@@ -268,14 +270,89 @@ 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_reads_default_kvs_directory_and_rejects_working_directory_file(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """Without an input record, the default store directory is read; a working-directory file next to it raises."""
     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"}')
+        with pytest.raises(RuntimeError, match='both in the default key-value store and in'):
+            await actor.get_input()
+
+
+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_conflicts_with_default_kvs_directory(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A working-directory file holding JSON `null` counts as present, so a default store directory file raises."""
+    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:
+        with pytest.raises(RuntimeError, match='both in the default key-value store and in'):
+            await actor.get_input()
+
+
+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)