feat(input_data): support file scheme in update

Author: yuji38kwmt

annofabcli/input_data/update_input_data.py

@@ -25,6 +25,7 @@
     get_json_from_args,
 )
 from annofabcli.common.facade import AnnofabApiFacade
+from annofabcli.common.utils import get_file_scheme_path
 
 logger = logging.getLogger(__name__)
 
@@ -81,11 +82,18 @@ def update_input_data(
             logger.warning(f"{log_prefix}入力データは存在しません。")
             return UpdateResult.SKIPPED
 
+        file_path = get_file_scheme_path(new_input_data_path) if new_input_data_path is not None else None
+        if file_path is not None and not Path(file_path).exists():
+            logger.warning(f"{log_prefix}input_data_path='{new_input_data_path}'にファイルは存在しません。入力データの更新をスキップします。")
+            return UpdateResult.SKIPPED
+
         # 更新する内容の確認メッセージを作成
         changes = []
         if new_input_data_name is not None:
             changes.append(f"input_data_name='{old_input_data['input_data_name']}'を'{new_input_data_name}'に変更")
-        if new_input_data_path is not None:
+        if file_path is not None:
+            changes.append(f"input_data_pathをローカルファイル'{file_path}'のアップロード結果に変更")
+        elif new_input_data_path is not None:
             changes.append(f"input_data_path='{old_input_data['input_data_path']}'を'{new_input_data_path}'に変更")
 
         if len(changes) == 0:
@@ -96,15 +104,18 @@ def update_input_data(
         if not self.confirm_processing(f"{log_prefix}{change_message}しますか？"):
             return UpdateResult.SKIPPED
 
-        request_body = old_input_data
+        request_body = dict(old_input_data)
         request_body["last_updated_datetime"] = old_input_data["updated_datetime"]
 
         if new_input_data_name is not None:
             request_body["input_data_name"] = new_input_data_name
-        if new_input_data_path is not None:
-            request_body["input_data_path"] = new_input_data_path
+        if file_path is not None:
+            self.service.wrapper.put_input_data_from_file(project_id, input_data_id=input_data_id, file_path=file_path, request_body=request_body)
+        else:
+            if new_input_data_path is not None:
+                request_body["input_data_path"] = new_input_data_path
+            self.service.api.put_input_data(project_id, input_data_id, request_body=request_body)
 
-        self.service.api.put_input_data(project_id, input_data_id, request_body=request_body)
         logger.debug(f"{log_prefix} :: 入力データを更新しました。 :: {changes}")
         return UpdateResult.SUCCESS
 
@@ -269,12 +280,12 @@ def parse_args(parser: argparse.ArgumentParser) -> None:
             " * ヘッダ行あり, カンマ区切り\n"
             " * input_data_id (required)\n"
             " * input_data_name (optional)\n"
-            " * input_data_path (optional)\n"
+            " * input_data_path (optional): ``file://`` を先頭に付けると、ローカルファイルを入力データに使用します。\n"
             "更新しないプロパティは、セルの値を空欄にしてください。\n"
         ),
     )
 
-    JSON_SAMPLE = '[{"input_data_id":"id1","input_data_name":"new_name1"},{"input_data_id":"id2","input_data_path":"new_path2"}]'  # noqa: N806
+    JSON_SAMPLE = '[{"input_data_id":"id1","input_data_name":"new_name1"},{"input_data_id":"id2","input_data_path":"file://new_image.jpg"}]'  # noqa: N806
     file_group.add_argument(
         "--json",
         type=str,

docs/command_reference/input_data/update.rst

@@ -27,7 +27,7 @@ CSVのフォーマットは以下の通りです。
 
     input_data_id,Yes,更新対象の入力データID
     input_data_name,No,変更後の入力データ名。更新しない場合は空欄
-    input_data_path,No,変更後の入力データパス。更新しない場合は空欄
+    input_data_path,No,変更後の入力データパス。更新しない場合は空欄。先頭が ``file://`` の場合、ローカルのファイルを入力データに使用します。
 
 
 以下はCSVファイルのサンプルです。
@@ -39,6 +39,7 @@ CSVのフォーマットは以下の通りです。
     id1,new_name1,
     id2,,s3://bucket/new_image.jpg
     id3,new_name3,https://example.com/new_image.jpg
+    id4,,file://new_image.jpg
 
 
 .. warning::
@@ -47,6 +48,8 @@ CSVのフォーマットは以下の通りです。
     プライベートストレージを利用するには、Annofabサポート窓口への問い合わせが必要です。
     詳細は https://annofab.readme.io/docs/external-storage を参照してください。
 
+``input_data_path`` の先頭が ``file://`` の場合、指定したローカルファイルをアップロードし、入力データのパスをアップロード後のパスに更新します。
+
 
 
 .. code-block::
@@ -78,6 +81,10 @@ JSON文字列を指定する場合
             "input_data_id": "id3",
             "input_data_name": "new_name3",
             "input_data_path": "https://example.com/new_image.jpg"
+        },
+        {
+            "input_data_id": "id4",
+            "input_data_path": "file://new_image.jpg"
         }
     ]
 

tests/input_data/test_update_input_data.py

@@ -0,0 +1,81 @@
+from pathlib import Path
+from types import SimpleNamespace
+from typing import cast
+from unittest.mock import Mock
+
+import annofabapi
+
+from annofabcli.input_data.update_input_data import UpdateInputDataMain, UpdateResult
+
+
+def create_old_input_data() -> dict[str, str]:
+    return {
+        "input_data_id": "input1",
+        "input_data_name": "old_name",
+        "input_data_path": "s3://bucket/old_image.jpg",
+        "updated_datetime": "2026-01-01T00:00:00.000+09:00",
+    }
+
+
+def create_service() -> tuple[annofabapi.Resource, Mock, Mock]:
+    api = Mock()
+    wrapper = Mock()
+    wrapper.get_input_data_or_none.return_value = create_old_input_data()
+    service = cast(annofabapi.Resource, SimpleNamespace(api=api, wrapper=wrapper))
+    return service, api, wrapper
+
+
+def test_update_input_data_with_url_path() -> None:
+    service, api, wrapper = create_service()
+    main_obj = UpdateInputDataMain(service, all_yes=True)
+
+    result = main_obj.update_input_data("project1", "input1", new_input_data_path="s3://bucket/new_image.jpg")
+
+    assert result == UpdateResult.SUCCESS
+    api.put_input_data.assert_called_once_with(
+        "project1",
+        "input1",
+        request_body={
+            "input_data_id": "input1",
+            "input_data_name": "old_name",
+            "input_data_path": "s3://bucket/new_image.jpg",
+            "updated_datetime": "2026-01-01T00:00:00.000+09:00",
+            "last_updated_datetime": "2026-01-01T00:00:00.000+09:00",
+        },
+    )
+    wrapper.put_input_data_from_file.assert_not_called()
+
+
+def test_update_input_data_with_file_scheme_path(tmp_path: Path) -> None:
+    input_data_file = tmp_path / "new_image.jpg"
+    input_data_file.write_bytes(b"image")
+    service, api, wrapper = create_service()
+    main_obj = UpdateInputDataMain(service, all_yes=True)
+
+    result = main_obj.update_input_data("project1", "input1", new_input_data_name="new_name", new_input_data_path=f"file://{input_data_file}")
+
+    assert result == UpdateResult.SUCCESS
+    wrapper.put_input_data_from_file.assert_called_once_with(
+        "project1",
+        input_data_id="input1",
+        file_path=str(input_data_file),
+        request_body={
+            "input_data_id": "input1",
+            "input_data_name": "new_name",
+            "input_data_path": "s3://bucket/old_image.jpg",
+            "updated_datetime": "2026-01-01T00:00:00.000+09:00",
+            "last_updated_datetime": "2026-01-01T00:00:00.000+09:00",
+        },
+    )
+    api.put_input_data.assert_not_called()
+
+
+def test_update_input_data_with_missing_file_scheme_path(tmp_path: Path) -> None:
+    service, api, wrapper = create_service()
+    main_obj = UpdateInputDataMain(service, all_yes=True)
+
+    result = main_obj.update_input_data("project1", "input1", new_input_data_path=f"file://{tmp_path / 'missing.jpg'}")
+
+    assert result == UpdateResult.SKIPPED
+    wrapper.put_input_data_from_file.assert_not_called()
+    api.put_input_data.assert_not_called()