Merge pull request #137 from kurusugawa-computer/modify-upload-file

file objectをs3にアップロードするメソッドを追加

Author: yuji38kwmt

annofabapi/__version__.py

@@ -1 +1 @@
-__version__ = '0.27.9'
+__version__ = '0.27.10'

annofabapi/api.py

@@ -1,7 +1,6 @@
 import functools
 import json
 import logging
-import warnings
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union  # pylint: disable=unused-import
 
 import backoff
@@ -87,6 +86,9 @@ def __init__(self, login_user_id: str, login_password: str, endpoint_url: str =
     #: login, refresh_tokenで取得したtoken情報
     token_dict: Optional[Dict[str, Any]] = None
 
+    #: Signed Cookie情報
+    cookies: Optional[Dict[str, Any]] = None
+
     class __MyToken(AuthBase):
         """
         requestsモジュールのauthに渡す情報。
@@ -106,6 +108,7 @@ def _create_kwargs(self, params: Optional[Dict[str, Any]] = None, headers: Optio
                        request_body: Optional[Any] = None) -> Dict[str, Any]:
         """
         requestsモジュールのget,...メソッドに渡すkwargsを生成する。
+
         Args:
             params: クエリパラメタに設定する情報
             headers: リクエストヘッダに設定する情報
@@ -211,26 +214,48 @@ def _request_wrapper(self, http_method: str, url_path: str, query_params: Option
         content = self._response_to_content(response)
         return content, response
 
-    def _get_signed_cookie(self, project_id):
+    def _get_signed_cookie(self, project_id) -> Tuple[Dict[str, Any], requests.Response]:
         """
         アノテーション仕様の履歴情報を取得するために、非公開APIにアクセスする。
         変更される可能性あり.
 
-        .. deprecated:: X
-
         Args:
             project_id: プロジェクトID
 
         Returns:
-            Tuple[Content, Reponse)
+            Tuple[Content, Response)
 
         """
-        warnings.warn("deprecated", DeprecationWarning)
         url_path = f'/private/projects/{project_id}/sign-headers'
         http_method = 'GET'
         keyword_params: Dict[str, Any] = {}
         return self._request_wrapper(http_method, url_path, **keyword_params)
 
+    def _request_get_with_cookie(self, project_id: str, url: str) -> requests.Response:
+        """
+        Signed Cookie を使って、AnnoFabのURLにGET requestを投げる。
+
+        Args:
+            project_id: プロジェクトID
+            url: アクセス対象のURL
+
+        Returns:
+            Response
+
+        """
+        if self.cookies is None:
+            self.cookies, _ = self._get_signed_cookie(project_id)
+
+        kwargs = {"cookies": self.cookies}
+        response = requests.get(url, **kwargs)
+
+        # CloudFrontから403 Errorが発生したとき
+        if response.status_code == requests.codes.forbidden and response.headers.get("server") == "CloudFront":
+            self.cookies, _ = self._get_signed_cookie(project_id)
+            return self._request_get_with_cookie(project_id, url)
+        else:
+            return response
+
     #########################################
     # Public Method : Login
     #########################################

annofabapi/wrapper.py

@@ -6,7 +6,7 @@
 import urllib
 import urllib.parse
 import warnings
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional
 
 import requests
 
@@ -241,14 +241,26 @@ def upload_file_to_s3(self, project_id: str, file_path: str, content_type: Optio
             content_type: アップロードするファイルのMIME Type. Noneの場合、ファイルパスから推測する。
 
         Returns:
-            AnnoFabに登録するときのpath
+            一時データ保存先であるS3パス
         """
 
         # content_type を推測
         new_content_type = self._get_content_type(file_path, content_type)
-        return self._upload_file_to_s3(project_id, file_path, content_type=new_content_type)
+        with open(file_path, 'rb') as f:
+            return self.upload_file_object_to_s3(project_id, fp=f, content_type=new_content_type)
+
+    def upload_file_object_to_s3(self, project_id: str, fp: typing.IO, content_type: str) -> str:
+        """
+        createTempPath APIを使ってアップロード用のURLとS3パスを取得して、"file object"をアップロードする。
+
+        Args:
+            project_id: プロジェクトID
+            fp: アップロードするファイルのfile object
+            content_type: アップロードするfile objectのMIME Type.
 
-    def _upload_file_to_s3(self, project_id: str, fp: Union[str, typing.IO], content_type: str) -> str:
+        Returns:
+            一時データ保存先であるS3パス
+        """
         # 一時データ保存先を取得
         content = self.api.create_temp_path(project_id, header_params={'content-type': content_type})[0]
 
@@ -259,12 +271,7 @@ def _upload_file_to_s3(self, project_id: str, fp: Union[str, typing.IO], content
         s3_url = content["url"].split("?")[0]
 
         # アップロード
-        if isinstance(fp, str):
-            with open(fp, 'rb') as f:
-                res_put = self.api.session.put(s3_url, params=query_dict, data=f,
-                                               headers={'content-type': content_type})
-        else:
-            res_put = self.api.session.put(s3_url, params=query_dict, data=fp, headers={'content-type': content_type})
+        res_put = self.api.session.put(s3_url, params=query_dict, data=fp, headers={'content-type': content_type})
 
         annofabapi.utils.log_error_response(logger, res_put)
         annofabapi.utils.raise_for_status(res_put)
@@ -818,7 +825,7 @@ def get_latest_instruction(self, project_id: str) -> Optional[Instruction]:
     def upload_instruction_image(self, project_id: str, image_id: str, file_path: str,
                                  content_type: Optional[str] = None) -> str:
         """
-        作業ガイドの画像をアップロードする。image_idはUUIDv4
+        作業ガイドの画像をアップロードする。
 
         Args:
             project_id: プロジェクトID
@@ -827,15 +834,30 @@ def upload_instruction_image(self, project_id: str, image_id: str, file_path: st
             content_type: アップロードするファイルのMIME Type. Noneの場合、ファイルパスから推測する。
 
         Returns:
-            AnnoFabに登録するときのpath
+            一時データ保存先であるS3パス
         """
-
-        # content_type を推測
         new_content_type = self._get_content_type(file_path, content_type)
+        with open(file_path, 'rb') as f:
+            return self.upload_file_object_as_instruction_image(project_id, image_id, fp=f,
+                                                                content_type=new_content_type)
+
+    def upload_file_object_as_instruction_image(self, project_id: str, image_id: str, fp: typing.IO,
+                                                content_type: str) -> str:
+        """
+        file objectを作業ガイドの画像としてアップロードする。
+
+        Args:
+            project_id: プロジェクトID
+            image_id: 作業ガイド画像ID
+            fp: アップロードするファイルのfile object
+            content_type: アップロードするファイルのMIME Type.
 
+        Returns:
+            一時データ保存先であるS3パス
+        """
         # 作業ガイド登録用/更新用のURLを取得
         content = self.api.get_instruction_image_url_for_put(project_id, image_id,
-                                                             header_params={'content-type': new_content_type})[0]
+                                                             header_params={'content-type': content_type})[0]
 
         url_parse_result = urllib.parse.urlparse(content["url"])
         query_dict = urllib.parse.parse_qs(url_parse_result.query)
@@ -844,9 +866,7 @@ def upload_instruction_image(self, project_id: str, image_id: str, file_path: st
         s3_url = content["url"].split("?")[0]
 
         # アップロード
-        with open(file_path, 'rb') as f:
-            res_put = self.api.session.put(s3_url, params=query_dict, data=f,
-                                           headers={'content-type': new_content_type})
+        res_put = self.api.session.put(s3_url, params=query_dict, data=fp, headers={'content-type': content_type})
         annofabapi.utils.log_error_response(logger, res_put)
         annofabapi.utils.raise_for_status(res_put)
         return content["path"]

tests/test_api.py

@@ -449,3 +449,11 @@ def test_get_task_or_none(self):
         assert wrapper.get_task_or_none(project_id, "not-exists") is None
 
         assert wrapper.get_task_or_none("not-exists", task_id) is None
+
+
+class TestProtectedMethod:
+    def test__request_get_with_cookie(self):
+        images, _ = api.get_instruction_images(project_id)
+        url = images[0]["url"]
+        r = api._request_get_with_cookie(project_id, url)
+        assert r.headers["Content-Type"].startswith("image/")