diff --git a/annofabapi/project_member_repository.py b/annofabapi/project_member_repository.py new file mode 100644 index 00000000..f7832ed3 --- /dev/null +++ b/annofabapi/project_member_repository.py @@ -0,0 +1,98 @@ +from collections.abc import Callable + +import more_itertools + +from annofabapi import Resource +from annofabapi.models import ProjectMember + + +class ProjectMemberRepository: + """プロジェクトメンバ情報を取得するRepository。""" + + def __init__(self, resource: Resource) -> None: + self.resource = resource + self._members_by_project_id: dict[str, list[ProjectMember]] = {} + + def _get_project_member_with_predicate(self, project_id: str, predicate: Callable[[ProjectMember], bool]) -> ProjectMember | None: + """条件に一致するプロジェクトメンバを取得する。 + + プロジェクトメンバの一覧をプロジェクトIDごとにキャッシュする。 + + Args: + project_id: プロジェクトID + predicate: プロジェクトメンバの検索条件 + + Returns: + 条件に一致するプロジェクトメンバ。見つからない場合はNone。 + """ + project_member_list = self._members_by_project_id.get(project_id) + if project_member_list is None: + project_member_list = self.resource.wrapper.get_all_project_members(project_id, query_params={"include_inactive_member": True}) + self._members_by_project_id[project_id] = project_member_list + return more_itertools.first_true(project_member_list, pred=predicate) + + def get_project_member_from_account_id(self, project_id: str, account_id: str) -> ProjectMember: + """account_idからプロジェクトメンバを取得する。 + + Args: + project_id: プロジェクトID + account_id: アカウントID + + Returns: + 指定したaccount_idのプロジェクトメンバ。 + + Raises: + ValueError: 指定したaccount_idのプロジェクトメンバが見つからない場合。 + """ + member = self._get_project_member_with_predicate(project_id, predicate=lambda e: e["account_id"] == account_id) + if member is None: + raise ValueError(f"project_member is not found. project_id='{project_id}', account_id='{account_id}'") + return member + + def get_project_member_from_user_id(self, project_id: str, user_id: str) -> ProjectMember: + """user_idからプロジェクトメンバを取得する。 + + Args: + project_id: プロジェクトID + user_id: ユーザーID + + Returns: + 指定したuser_idのプロジェクトメンバ。 + + Raises: + ValueError: 指定したuser_idのプロジェクトメンバが見つからない場合。 + """ + member = self._get_project_member_with_predicate(project_id, predicate=lambda e: e["user_id"] == user_id) + if member is None: + raise ValueError(f"project_member is not found. project_id='{project_id}', user_id='{user_id}'") + return member + + def get_user_id_from_account_id(self, project_id: str, account_id: str) -> str: + """account_idからuser_idを取得する。 + + Args: + project_id: プロジェクトID + account_id: アカウントID + + Returns: + 指定したaccount_idに対応するユーザーID。 + + Raises: + ValueError: 指定したaccount_idのプロジェクトメンバが見つからない場合。 + """ + return self.get_project_member_from_account_id(project_id, account_id)["user_id"] + + def get_account_id_from_user_id(self, project_id: str, user_id: str) -> str: + """user_idからaccount_idを取得する。 + + Args: + project_id: プロジェクトID + user_id: ユーザーID + + Returns: + 指定したuser_idに対応するアカウントID。 + + Raises: + ValueError: 指定したuser_idのプロジェクトメンバが見つからない場合。 + """ + return self.get_project_member_from_user_id(project_id, user_id)["account_id"] diff --git a/annofabapi/util/annotation_specs.py b/annofabapi/util/annotation_specs.py index 749f1078..5740f781 100644 --- a/annofabapi/util/annotation_specs.py +++ b/annofabapi/util/annotation_specs.py @@ -1,4 +1,5 @@ -from typing import Any, Literal, TypedDict, cast +from collections.abc import Mapping +from typing import Any, Literal, TypedDict import more_itertools @@ -43,7 +44,19 @@ class LabelDefinition(TypedDict): additional_data_definitions: list[str] -def get_english_message(internationalization_message: InternationalizationMessage | dict[str, Any]) -> str: +class LabelNameHolder(TypedDict): + """ラベル名を持つ情報です。""" + + label_name: InternationalizationMessage + + +class NameHolder(TypedDict): + """多言語化された名前を持つ情報です。""" + + name: InternationalizationMessage + + +def get_english_message(internationalization_message: Mapping[str, Any]) -> str: """ `InternationalizationMessage`クラスの値から、英語メッセージを取得します。 英語メッセージが見つからない場合は ``ValueError`` をスローします。 @@ -60,7 +73,7 @@ def get_english_message(internationalization_message: InternationalizationMessag Raises: ValueError: 英語メッセージが見つからない場合 """ - messages = cast(list[InternationalizationMessageItem], internationalization_message["messages"]) + messages = internationalization_message["messages"] result = more_itertools.first_true(messages, pred=lambda e: e["lang"] == Lang.EN_US.value) if result is not None: return result["message"] @@ -98,6 +111,54 @@ def get_message_with_lang(internationalization_message: InternationalizationMess return None +def get_label_name_en(label: Mapping[str, Any]) -> str: + """ + ラベル情報から英語名を取得します。 + + Args: + label: ラベル情報。キー ``label_name`` が存在している必要があります。 + + Returns: + ラベルの英語名。 + + Raises: + ValueError: 英語メッセージが見つからない場合 + """ + return get_english_message(label["label_name"]) + + +def get_attribute_name_en(attribute: Mapping[str, Any]) -> str: + """ + 属性情報から英語名を取得します。 + + Args: + attribute: 属性情報。キー ``name`` が存在している必要があります。 + + Returns: + 属性の英語名。 + + Raises: + ValueError: 英語メッセージが見つからない場合 + """ + return get_english_message(attribute["name"]) + + +def get_choice_name_en(choice: Mapping[str, Any]) -> str: + """ + 選択肢情報から英語名を取得します。 + + Args: + choice: 選択肢情報。キー ``name`` が存在している必要があります。 + + Returns: + 選択肢の英語名。 + + Raises: + ValueError: 英語メッセージが見つからない場合 + """ + return get_english_message(choice["name"]) + + def get_choice(choices: list[AttributeChoice], *, choice_id: str | None = None, choice_name: str | None = None) -> AttributeChoice: """ 選択肢情報を取得します。 @@ -116,7 +177,7 @@ def get_choice(choices: list[AttributeChoice], *, choice_id: str | None = None, if choice_id is not None: result = [e for e in choices if e["choice_id"] == choice_id] elif choice_name is not None: - result = [e for e in choices if get_english_message(e["name"]) == choice_name] + result = [e for e in choices if get_choice_name_en(e) == choice_name] else: raise ValueError("'choice_id'か'choice_name'のどちらかはNone以外にしてください。") @@ -151,14 +212,14 @@ def get_attribute( if attribute_id is not None: result = [e for e in additionals if e["additional_data_definition_id"] == attribute_id] elif attribute_name is not None: - result = [e for e in additionals if get_english_message(e["name"]) == attribute_name] + result = [e for e in additionals if get_attribute_name_en(e) == attribute_name] else: raise ValueError("'attribute_id'か'attribute_name'のどちらかはNone以外にしてください。") label_name = None if label is not None: result = [e for e in result if e["additional_data_definition_id"] in label["additional_data_definitions"]] - label_name = get_english_message(label["label_name"]) + label_name = get_label_name_en(label) if len(result) == 0: raise ValueError( @@ -189,7 +250,7 @@ def get_label(labels: list[LabelDefinition], *, label_id: str | None = None, lab if label_id is not None: result = [e for e in labels if e["label_id"] == label_id] elif label_name is not None: - result = [e for e in labels if get_english_message(e["label_name"]) == label_name] + result = [e for e in labels if get_label_name_en(e) == label_name] else: raise ValueError("'label_id'か'label_name'のどちらかはNone以外にしてください。") diff --git a/annofabapi/wrapper.py b/annofabapi/wrapper.py index 20eada09..e2a91df7 100644 --- a/annofabapi/wrapper.py +++ b/annofabapi/wrapper.py @@ -47,6 +47,7 @@ TaskStatus, ) from annofabapi.parser import SimpleAnnotationDirParser, SimpleAnnotationParser +from annofabapi.util.annotation_specs import get_attribute_name_en, get_choice_name_en, get_label_name_en logger = logging.getLogger(__name__) @@ -533,19 +534,19 @@ def copy_annotation( def __get_label_info_from_label_name(self, label_name: str, annotation_specs_labels: list[LabelV1]) -> LabelV1 | None: for label in annotation_specs_labels: - if self.__get_label_name_en(label) == label_name: + if get_label_name_en(label) == label_name: return label return None def __get_additional_data_from_attribute_name(self, attribute_name: str, label_info: LabelV1) -> AdditionalDataDefinitionV1 | None: for additional_data in label_info["additional_data_definitions"]: - if self.__get_additional_data_definition_name_en(additional_data) == attribute_name: + if get_attribute_name_en(additional_data) == attribute_name: return additional_data return None def _get_choice_id_from_name(self, name: str, choices: list[dict[str, Any]]) -> str | None: - choice_info = more_itertools.first_true(choices, pred=lambda e: self.__get_choice_name_en(e) == name) + choice_info = more_itertools.first_true(choices, pred=lambda e: get_choice_name_en(e) == name) if choice_info is not None: return choice_info["choice_id"] else: @@ -572,7 +573,7 @@ def __to_additional_data_list(self, attributes: dict[str, Any], label_info: Labe if specs_additional_data is None: logger.warning( "アノテーション仕様の '%s' ラベルに、attribute_name='%s' である属性が存在しません。", - self.__get_label_name_en(label_info), + get_label_name_en(label_info), key, ) continue @@ -698,7 +699,7 @@ def to_label_v1(label_v2: dict[str, Any]) -> LabelV1: else: raise ValueError( f"additional_data_definition_id='{additional_data_definition_id}' に対応する属性情報が存在しません。" - f"label_id='{label_v2['label_id']}', label_name_en='{self.__get_label_name_en(label_v2)}'" + f"label_id='{label_v2['label_id']}', label_name_en='{get_label_name_en(label_v2)}'" ) label_v2["additional_data_definitions"] = new_additional_data_definitions return label_v2 @@ -784,24 +785,6 @@ def put_annotation_for_simple_annotation_json( # Public Method : AnnotationSpecs ######################################### - @staticmethod - def __get_label_name_en(label: dict[str, Any]) -> str: - """label情報から英語名を取得する""" - label_name_messages = label["label_name"]["messages"] - return next(e["message"] for e in label_name_messages if e["lang"] == "en-US") - - @staticmethod - def __get_additional_data_definition_name_en(additional_data_definition: dict[str, Any]) -> str: - """additional_data_definitionから英語名を取得する""" - messages = additional_data_definition["name"]["messages"] - return next(e["message"] for e in messages if e["lang"] == "en-US") - - @staticmethod - def __get_choice_name_en(choice: dict[str, Any]) -> str: - """choiceから英語名を取得する""" - messages = choice["name"]["messages"] - return next(e["message"] for e in messages if e["lang"] == "en-US") - def __get_dest_additional( self, src_additional: dict[str, Any], @@ -810,9 +793,9 @@ def __get_dest_additional( dest_labels: list[dict[str, Any]], dict_label_id: dict[str, str], ) -> dict[str, Any] | None: - src_additional_name_en = self.__get_additional_data_definition_name_en(src_additional) + src_additional_name_en = get_attribute_name_en(src_additional) for dest_additional in dest_additionals: - if src_additional_name_en != self.__get_additional_data_definition_name_en(dest_additional): + if src_additional_name_en != get_attribute_name_en(dest_additional): continue dest_label_contains_dest_additional = True @@ -861,10 +844,10 @@ def get_annotation_specs_relation(self, src_project_id: str, dest_project_id: st dict_label_id: dict[str, str] = {} for src_label in src_annotation_specs["labels"]: - src_label_name_en = self.__get_label_name_en(src_label) + src_label_name_en = get_label_name_en(src_label) dest_label = more_itertools.first_true( dest_labels, - pred=lambda e: self.__get_label_name_en(e) == src_label_name_en, # pylint: disable=cell-var-from-loop # noqa: B023 + pred=lambda e: get_label_name_en(e) == src_label_name_en, # pylint: disable=cell-var-from-loop # noqa: B023 ) if dest_label is not None: dict_label_id[src_label["label_id"]] = dest_label["label_id"] @@ -886,10 +869,10 @@ def get_annotation_specs_relation(self, src_project_id: str, dest_project_id: st dest_choices = dest_additional["choices"] for src_choice in src_additional["choices"]: - src_choice_name_en = self.__get_choice_name_en(src_choice) + src_choice_name_en = get_choice_name_en(src_choice) dest_choice = more_itertools.first_true( dest_choices, - pred=lambda e: self.__get_choice_name_en(e) == src_choice_name_en, # pylint: disable=cell-var-from-loop # noqa: B023 + pred=lambda e: get_choice_name_en(e) == src_choice_name_en, # pylint: disable=cell-var-from-loop # noqa: B023 ) if dest_choice is not None: dict_choice_id[ChoiceKey(src_additional["additional_data_definition_id"], src_choice["choice_id"])] = ChoiceKey( diff --git a/docs/api_reference/index.rst b/docs/api_reference/index.rst index 4595c2e3..9babc7e0 100644 --- a/docs/api_reference/index.rst +++ b/docs/api_reference/index.rst @@ -9,14 +9,12 @@ API reference wrapper resource parser - plugin dataclass exceptions segmentation - pydantic_models - models + plugin + project_member_repository util utils - credentials - - + pydantic_models + models diff --git a/docs/api_reference/project_member_repository.rst b/docs/api_reference/project_member_repository.rst new file mode 100644 index 00000000..a447856e --- /dev/null +++ b/docs/api_reference/project_member_repository.rst @@ -0,0 +1,7 @@ +annofabapi.project_member_repository module +=========================================== + + + +.. automodule:: annofabapi.project_member_repository + :members: diff --git a/tests/test_project_member_repository.py b/tests/test_project_member_repository.py new file mode 100644 index 00000000..455b099d --- /dev/null +++ b/tests/test_project_member_repository.py @@ -0,0 +1,89 @@ +import pytest + +from annofabapi.project_member_repository import ProjectMemberRepository + +PROJECT_ID = "project_id" + + +def create_repository_with_members(members): + repository = object.__new__(ProjectMemberRepository) + repository._members_by_project_id = {PROJECT_ID: members} + return repository + + +def test_get_user_id_from_account_id__存在するaccount_idならuser_idを返す(): + repository = create_repository_with_members( + [ + { + "project_id": PROJECT_ID, + "account_id": "account_id", + "user_id": "user_id", + } + ] + ) + + assert repository.get_user_id_from_account_id(PROJECT_ID, "account_id") == "user_id" + + +def test_get_project_member_from_account_id__存在するaccount_idならプロジェクトメンバを返す(): + member = { + "project_id": PROJECT_ID, + "account_id": "account_id", + "user_id": "user_id", + } + repository = create_repository_with_members([member]) + + assert repository.get_project_member_from_account_id(PROJECT_ID, "account_id") == member + + +def test_get_project_member_from_account_id__存在しないaccount_idならValueErrorを送出する(): + repository = create_repository_with_members([]) + + with pytest.raises(ValueError): + repository.get_project_member_from_account_id(PROJECT_ID, "unknown_account_id") + + +def test_get_user_id_from_account_id__存在しないaccount_idならValueErrorを送出する(): + repository = create_repository_with_members([]) + + with pytest.raises(ValueError): + repository.get_user_id_from_account_id(PROJECT_ID, "unknown_account_id") + + +def test_get_account_id_from_user_id__存在するuser_idならaccount_idを返す(): + repository = create_repository_with_members( + [ + { + "project_id": PROJECT_ID, + "account_id": "account_id", + "user_id": "user_id", + } + ] + ) + + assert repository.get_account_id_from_user_id(PROJECT_ID, "user_id") == "account_id" + + +def test_get_project_member_from_user_id__存在するuser_idならプロジェクトメンバを返す(): + member = { + "project_id": PROJECT_ID, + "account_id": "account_id", + "user_id": "user_id", + } + repository = create_repository_with_members([member]) + + assert repository.get_project_member_from_user_id(PROJECT_ID, "user_id") == member + + +def test_get_project_member_from_user_id__存在しないuser_idならValueErrorを送出する(): + repository = create_repository_with_members([]) + + with pytest.raises(ValueError): + repository.get_project_member_from_user_id(PROJECT_ID, "unknown_user_id") + + +def test_get_account_id_from_user_id__存在しないuser_idならValueErrorを送出する(): + repository = create_repository_with_members([]) + + with pytest.raises(ValueError): + repository.get_account_id_from_user_id(PROJECT_ID, "unknown_user_id") diff --git a/tests/util/test_annotation_specs.py b/tests/util/test_annotation_specs.py index 31b8a22a..1d30baca 100644 --- a/tests/util/test_annotation_specs.py +++ b/tests/util/test_annotation_specs.py @@ -1,6 +1,18 @@ import pytest -from annofabapi.util.annotation_specs import AnnotationSpecsAccessor, AttributeChoice, Lang, get_choice, get_english_message, get_message_with_lang +from annofabapi.util.annotation_specs import ( + AnnotationSpecsAccessor, + AttributeChoice, + LabelNameHolder, + Lang, + NameHolder, + get_attribute_name_en, + get_choice, + get_choice_name_en, + get_english_message, + get_label_name_en, + get_message_with_lang, +) class Test__get_english_message: @@ -31,6 +43,57 @@ def test__get_message_with_lang(self): assert get_message_with_lang(i18n_message, Lang.VI_VN) is None # type: ignore[arg-type] +class Test__get_label_name_en: + def test__get_label_name_en(self): + label: LabelNameHolder = { + "label_name": {"messages": [{"lang": "ja-JP", "message": "自動車"}, {"lang": "en-US", "message": "Car"}]}, + } + + assert get_label_name_en(label) == "Car" + + def test__get_label_name_en__英語メッセージが存在しない場合はValueErrorをスローする(self): + label: LabelNameHolder = { + "label_name": {"messages": [{"lang": "ja-JP", "message": "自動車"}]}, + } + + with pytest.raises(ValueError): + get_label_name_en(label) + + +class Test__get_attribute_name_en: + def test__get_attribute_name_en(self): + attribute: NameHolder = { + "name": {"messages": [{"lang": "ja-JP", "message": "色"}, {"lang": "en-US", "message": "Color"}]}, + } + + assert get_attribute_name_en(attribute) == "Color" + + def test__get_attribute_name_en__英語メッセージが存在しない場合はValueErrorをスローする(self): + attribute: NameHolder = { + "name": {"messages": [{"lang": "ja-JP", "message": "色"}]}, + } + + with pytest.raises(ValueError): + get_attribute_name_en(attribute) + + +class Test__get_choice_name_en: + def test__get_choice_name_en(self): + choice: NameHolder = { + "name": {"messages": [{"lang": "ja-JP", "message": "赤"}, {"lang": "en-US", "message": "Red"}]}, + } + + assert get_choice_name_en(choice) == "Red" + + def test__get_choice_name_en__英語メッセージが存在しない場合はValueErrorをスローする(self): + choice: NameHolder = { + "name": {"messages": [{"lang": "ja-JP", "message": "赤"}]}, + } + + with pytest.raises(ValueError): + get_choice_name_en(choice) + + class Test__AnnotationSpecsAccessor: def setup_method(self): self.annotation_specs = { @@ -48,12 +111,12 @@ def setup_method(self): def test_get_label_by_id(self): label = self.accessor.get_label(label_id="1") assert label["label_id"] == "1" - assert get_english_message(label["label_name"]) == "Car" + assert get_label_name_en(label) == "Car" def test_get_label_by_name(self): label = self.accessor.get_label(label_name="Bike") assert label["label_id"] == "2" - assert get_english_message(label["label_name"]) == "Bike" + assert get_label_name_en(label) == "Bike" def test_get_label_not_found(self): with pytest.raises(ValueError): @@ -62,12 +125,12 @@ def test_get_label_not_found(self): def test_get_attribute_by_id(self): attribute = self.accessor.get_attribute(attribute_id="1") assert attribute["additional_data_definition_id"] == "1" - assert get_english_message(attribute["name"]) == "Color" + assert get_attribute_name_en(attribute) == "Color" def test_get_attribute_by_name(self): attribute = self.accessor.get_attribute(attribute_name="Size") assert attribute["additional_data_definition_id"] == "2" - assert get_english_message(attribute["name"]) == "Size" + assert get_attribute_name_en(attribute) == "Size" def test_get_attribute_not_found(self): with pytest.raises(ValueError): @@ -77,7 +140,7 @@ def test_get_attribute_by_id_and_label(self): label = self.accessor.get_label(label_id="1") attribute = self.accessor.get_attribute(attribute_id="1", label=label) assert attribute["additional_data_definition_id"] == "1" - assert get_english_message(attribute["name"]) == "Color" + assert get_attribute_name_en(attribute) == "Color" def test_get_attribute_by_id_and_label__not_found(self): label = self.accessor.get_label(label_id="2") @@ -95,12 +158,12 @@ def setup_method(self): def test_get_choice_by_id(self): choice = get_choice(self.choices, choice_id="1") assert choice["choice_id"] == "1" - assert get_english_message(choice["name"]) == "Option1" + assert get_choice_name_en(choice) == "Option1" def test_get_choice_by_name(self): choice = get_choice(self.choices, choice_name="Option2") assert choice["choice_id"] == "2" - assert get_english_message(choice["name"]) == "Option2" + assert get_choice_name_en(choice) == "Option2" def test_get_choice_not_found(self): with pytest.raises(ValueError):