S3へのアップロードしたファイルが破損していないかのチェック (#360)

* md5 check

* test

* test

* update test

* bug fix

* コメントを追加

* CheckSumErrorというexceptionに変更

* format

* version up

Author: yuji38kwmt

annofabapi/__version__.py

@@ -1 +1 @@
-__version__ = "0.47.2"
+__version__ = "0.48.0"

annofabapi/exceptions.py

@@ -30,3 +30,23 @@ def __init__(self, outer_file_path: str, zipfile_path: Optional[str] = None):
             message = f"There is no item named '{str(outer_file_path)}' in the archive '{zipfile_path}'"
 
         super().__init__(message)
+
+
+class CheckSumError(AnnofabApiException):
+    """
+    アップロードしたデータ（ファイルやバイナリデータ）の整合性が一致していないときのエラー。
+
+    Args:
+        uploaded_data_hash: アップロード対象のデータのハッシュ値（MD5）
+        response_etag: アップロードしたときのレスポンスヘッダ'ETag'の値
+
+    Attributes:
+        uploaded_data_hash: アップロード対象のデータのハッシュ値（MD5）
+        response_etag: アップロードしたときのレスポンスヘッダ'ETag'の値
+    """
+
+    def __init__(self, message: str, uploaded_data_hash: str, response_etag: str):
+        self.uploaded_data_hash = uploaded_data_hash
+        self.response_etag = response_etag
+
+        super().__init__(message)

annofabapi/wrapper.py

@@ -2,6 +2,7 @@
 import asyncio
 import copy
 import datetime
+import hashlib
 import logging
 import mimetypes
 import time
@@ -16,7 +17,7 @@
 import requests
 
 from annofabapi import AnnofabApi
-from annofabapi.exceptions import AnnofabApiException
+from annofabapi.exceptions import AnnofabApiException, CheckSumError
 from annofabapi.models import (
     AdditionalData,
     AdditionalDataDefinitionType,
@@ -346,23 +347,38 @@ def __to_dest_annotation_detail(
     ) -> Dict[str, Any]:
         """
         コピー元の１個のアノテーションを、コピー先用に変換する。
-        塗りつぶし画像の場合、S3にアップロードする。
+        塗りつぶし画像などの外部アノテーションファイルがある場合、S3にアップロードする。
 
         Notes:
             annotation_id をUUIDv4で生成すると、アノテーションリンク属性をコピーしたときに対応できないので、暫定的にannotation_idは維持するようにする。
+
+        Raises:
+            CheckSumError: アップロードした外部アノテーションファイルのMD5ハッシュ値が、S3にアップロードしたときのレスポンスのETagに一致しない
+
         """
         dest_detail = detail
         dest_detail["account_id"] = account_id
         if detail["data_holding_type"] == AnnotationDataHoldingType.OUTER.value:
-            outer_file_url = detail["url"]
-            src_response = self._request_get_wrapper(outer_file_url)
-            s3_path = self.upload_data_to_s3(
-                dest_project_id, data=src_response.content, content_type=src_response.headers["Content-Type"]
-            )
-            logger.debug("%s に塗りつぶし画像をアップロードしました。", s3_path)
-            dest_detail["path"] = s3_path
-            dest_detail["url"] = None
-            dest_detail["etag"] = None
+
+            try:
+                outer_file_url = detail["url"]
+                src_response = self._request_get_wrapper(outer_file_url)
+                s3_path = self.upload_data_to_s3(
+                    dest_project_id, data=src_response.content, content_type=src_response.headers["Content-Type"]
+                )
+                logger.debug("%s に外部アノテーションファイルをアップロードしました。", s3_path)
+                dest_detail["path"] = s3_path
+                dest_detail["url"] = None
+                dest_detail["etag"] = None
+
+            except CheckSumError as e:
+                message = (
+                    f"外部アノテーションファイル {outer_file_url} のレスポンスのMD5ハッシュ値('{e.uploaded_data_hash}')が、"
+                    f"AWS S3にアップロードしたときのレスポンスのETag('{e.response_etag}')に一致しませんでした。アップロード時にデータが破損した可能性があります。"
+                )
+                raise CheckSumError(
+                    message=message, uploaded_data_hash=e.uploaded_data_hash, response_etag=e.response_etag
+                ) from e
 
         return dest_detail
 
@@ -533,6 +549,9 @@ def __to_annotation_detail_for_request(
 
         Returns:
 
+        Raises:
+            CheckSumError: アップロードした外部アノテーションファイルのMD5ハッシュ値が、S3にアップロードしたときのレスポンスのETagに一致しない
+
         """
         label_info = self.__get_label_info_from_label_name(detail["label"], annotation_specs_labels)
         if label_info is None:
@@ -559,10 +578,21 @@ def __to_annotation_detail_for_request(
 
         if data_holding_type == AnnotationDataHoldingType.OUTER.value:
             data_uri = detail["data"]["data_uri"]
+            outer_file_path = f"{parser.task_id}/{parser.input_data_id}/{data_uri}"
             with parser.open_outer_file(data_uri) as f:
-                s3_path = self.upload_data_to_s3(project_id, f, content_type="image/png")
-                dest_obj["path"] = s3_path
-                logger.debug(f"{parser.task_id}/{parser.input_data_id}/{data_uri} をS3にアップロードしました。")
+                try:
+                    s3_path = self.upload_data_to_s3(project_id, f, content_type="image/png")
+                    dest_obj["path"] = s3_path
+                    logger.debug(f"{outer_file_path} をS3にアップロードしました。")
+
+                except CheckSumError as e:
+                    message = (
+                        f"アップロードした外部アノテーションファイル'{outer_file_path}'のMD5ハッシュ値('{e.uploaded_data_hash}')が、"
+                        f"AWS S3にアップロードしたときのレスポンスのETag('{e.response_etag}')に一致しませんでした。アップロード時にデータが破損した可能性があります。"
+                    )
+                    raise CheckSumError(
+                        message=message, uploaded_data_hash=e.uploaded_data_hash, response_etag=e.response_etag
+                    ) from e
 
         return dest_obj
 
@@ -830,25 +860,51 @@ def upload_file_to_s3(self, project_id: str, file_path: str, content_type: Optio
 
         Returns:
             一時データ保存先であるS3パス
+
+        Raises:
+            CheckSumError: アップロードしたファイルのMD5ハッシュ値が、S3にアップロードしたときのレスポンスのETagと一致しない
+
         """
 
         # content_type を推測
         new_content_type = self._get_content_type(file_path, content_type)
         with open(file_path, "rb") as f:
-            return self.upload_data_to_s3(project_id, data=f, content_type=new_content_type)
+            try:
+                return self.upload_data_to_s3(project_id, data=f, content_type=new_content_type)
+            except CheckSumError as e:
+                message = (
+                    f"アップロードしたファイル'{file_path}'のMD5ハッシュ値('{e.uploaded_data_hash}')が、"
+                    f"AWS S3にアップロードしたときのレスポンスのETag('{e.response_etag}')に一致しませんでした。アップロード時にデータが破損した可能性があります。"
+                )
+                raise CheckSumError(
+                    message=message, uploaded_data_hash=e.uploaded_data_hash, response_etag=e.response_etag
+                ) from e
 
     def upload_data_to_s3(self, project_id: str, data: Any, content_type: str) -> str:
         """
         createTempPath APIを使ってアップロード用のURLとS3パスを取得して、"data" をアップロードする。
 
         Args:
             project_id: プロジェクトID
-            data: アップロード対象のdata. ``requests.put`` メソッドの ``data`` 引数にそのまま渡す。
+            data: アップロード対象のdata. ``open(mode="b")`` 関数の戻り値、またはバイナリ型の値です。 ``requests.put`` メソッドの ``data`` 引数にそのまま渡します。
             content_type: アップロードするfile objectのMIME Type.
 
         Returns:
             一時データ保存先であるS3パス
+
+        Raises:
+            CheckSumError: アップロードしたデータのMD5ハッシュ値が、S3にアップロードしたときのレスポンスのETagと一致しない
         """
+
+        def get_md5_value_from_file(fp):
+            md5_obj = hashlib.md5()
+            while True:
+                chunk = fp.read(2048 * md5_obj.block_size)
+                if len(chunk) == 0:
+                    break
+                md5_obj.update(chunk)
+            return md5_obj.hexdigest()
+
         # 一時データ保存先を取得
         content = self.api.create_temp_path(project_id, header_params={"content-type": content_type})[0]
 
@@ -865,6 +921,25 @@ def upload_data_to_s3(self, project_id: str, data: Any, content_type: str) -> st
 
         _log_error_response(logger, res_put)
         _raise_for_status(res_put)
+
+        # アップロードしたファイルが破損していなかをチェックする
+        if hasattr(data, "read"):
+            # 読み込み位置を先頭に戻す
+            data.seek(0)
+            uploaded_data_hash = get_md5_value_from_file(data)
+        else:
+            uploaded_data_hash = hashlib.md5(data).hexdigest()
+
+        # ETagにはダブルクォートが含まれているため、`str_md5`もそれに合わせる
+        response_etag = res_put.headers["ETag"]
+
+        if f'"{uploaded_data_hash}"' != response_etag:
+            message = (
+                f"アップロードしたデータのMD5ハッシュ値('{uploaded_data_hash}')が、"
+                f"AWS S3にアップロードしたときのレスポンスのETag('{response_etag}')に一致しませんでした。アップロード時にデータが破損した可能性があります。"
+            )
+            raise CheckSumError(message=message, uploaded_data_hash=uploaded_data_hash, response_etag=response_etag)
+
         return content["path"]
 
     def put_input_data_from_file(

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "annofabapi"
-version = "0.47.2"
+version = "0.48.0"
 description = "Python Clinet Library of AnnoFab WebAPI (https://annofab.com/docs/api/)"
 authors = ["yuji38kwmt"]
 license = "MIT"

pytest.ini

@@ -11,5 +11,8 @@ endpoint_url = https://annofab.com
 # Specify AnnoFab project that has owner role assigned to you.
 project_id = 1ae6ec18-2a71-4eb5-9ac1-92329b01a5ca
 task_id = test_task_1
+# 変更されるタスクのtask_id
+changed_task_id = test_task_2
+
 
 

tests/test_api.py

@@ -28,7 +28,7 @@
 
 project_id = inifile["annofab"]["project_id"]
 task_id = inifile["annofab"]["task_id"]
-
+changed_task_id = inifile["annofab"]["changed_task_id"]
 
 test_dir = "./tests/data"
 out_dir = "./tests/out"
@@ -84,6 +84,18 @@ def test_wrapper_copy_annotation(self):
         result = wrapper.copy_annotation(src, dest)
         assert result == True
 
+    def test_wrapper_put_annotation_for_simple_annotation_json(self):
+        input_data_id = test_wrapper.get_first_input_data_id_in_task(project_id, task_id)
+        annotation_specs_v2, _ = service.api.get_annotation_specs(project_id, query_params={"v": "2"})
+        wrapper.put_annotation_for_simple_annotation_json(
+            project_id,
+            changed_task_id,
+            input_data_id,
+            simple_annotation_json="tests/data/simple-annotation/sample_1/c6e1c2ec-6c7c-41c6-9639-4244c2ed2839.json",
+            annotation_specs_labels=annotation_specs_v2["labels"],
+            annotation_specs_additionals=annotation_specs_v2["additionals"],
+        )
+
 
 class TestAnnotationSpecs:
     def test_get_annotation_specs(self):
@@ -168,7 +180,7 @@ def teardown_class(cls):
         wrapper.change_task_status_to_break(project_id, task_id)
 
 
-class TestInput:
+class TestInputData:
     @classmethod
     def setup_class(cls):
         cls.input_data_id = test_wrapper.get_first_input_data_id_in_task(project_id, task_id)

tests/test_dataclass_webapi.py

@@ -1,6 +1,5 @@
 import configparser
 import os
-import warnings
 from pathlib import Path
 
 import annofabapi
@@ -122,7 +121,6 @@ def test_job(self):
             print(f"ジョブが存在しませんでした。")
 
 
-
 class TestMy:
     def test_my_organization(self):
         my_organizations = service.wrapper.get_all_my_organizations()