Enhance project member management and update documentation

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"]

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以外にしてください。")
 

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(

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

docs/api_reference/project_member_repository.rst

@@ -0,0 +1,7 @@
+annofabapi.project_member_repository module
+===========================================
+
+
+
+.. automodule:: annofabapi.project_member_repository
+    :members: