Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion annofabapi/__version__.py
Original file line number Diff line number Diff line change
@@ -1 +1 @@
__version__ = '0.27.9'
__version__ = '0.27.10'
37 changes: 31 additions & 6 deletions annofabapi/api.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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に渡す情報。
Expand All @@ -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: リクエストヘッダに設定する情報
Expand Down Expand Up @@ -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
#########################################
Expand Down
56 changes: 38 additions & 18 deletions annofabapi/wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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]

Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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"]
Expand Down
8 changes: 8 additions & 0 deletions tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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/")