Repository: gtalarico/pyairtable
Pull request #332 (2024-01-26 05:46:44)

Use improved enterprise "get users" endpoint, and simplify collaborations model
Two changes here:

1. Calling `enterprise.user()` or `enterprise.users()` now uses the [get users by id or email](https://airtable.com/developers/web/api/get-users-by-id-or-email) endpoint, which used to only accept emails (at least, according to its documentation). This removes the need to iterate over IDs and call [get user by id](https://airtable.com/developers/web/api/get-user-by-id) one by one.
2. When retrieving a user or group, we will now request collaborations by default. This can be disabled by passing `collaborations=False` as a keyword argument to `user()`, `users()`, or `group()`. This removes the need for type-checked code to do `if user.collaborations: ...` to check against `None`.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/gtalarico/pyairtable/pull/332?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`54ef64f`)](https://app.codecov.io/gh/gtalarico/pyairtable/commit/54ef64fb6eb56df1d91cef6e9d8a7f092cbb0e1d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico) 99.61% compared to head [(`d2f5311`)](https://app.codecov.io/gh/gtalarico/pyairtable/pull/332?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico) 99.61%.
> Report is 2 commits behind head on main.


<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##             main     #332   +/-   ##
=======================================
  Coverage   99.61%   99.61%           
=======================================
  Files          24       24           
  Lines        1801     1803    +2     
=======================================
+ Hits         1794     1796    +2     
  Misses          7        7           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/gtalarico/pyairtable/pull/332?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico).


Patch:
diff --git a/pyairtable/api/enterprise.py b/pyairtable/api/enterprise.py
index e772993..ae88108 100644
--- a/pyairtable/api/enterprise.py
+++ b/pyairtable/api/enterprise.py
@@ -1,7 +1,7 @@
-from typing import Dict, Iterable, List, Optional
+from typing import Iterable, List, Optional
 
 from pyairtable.models.schema import EnterpriseInfo, UserGroup, UserInfo
-from pyairtable.utils import cache_unless_forced, enterprise_only, is_user_id
+from pyairtable.utils import cache_unless_forced, enterprise_only
 
 
 @enterprise_only
@@ -32,61 +32,59 @@ class Enterprise:
         payload = self.api.request("GET", self.url, params=params)
         return EnterpriseInfo.parse_obj(payload)
 
-    def group(self, group_id: str) -> UserGroup:
+    def group(self, group_id: str, collaborations: bool = True) -> UserGroup:
+        params = {"include": ["collaborations"] if collaborations else []}
         url = self.api.build_url(f"meta/groups/{group_id}")
-        return UserGroup.parse_obj(self.api.request("GET", url))
+        payload = self.api.request("GET", url, params=params)
+        return UserGroup.parse_obj(payload)
 
-    def user(self, id_or_email: str) -> UserInfo:
+    def user(self, id_or_email: str, collaborations: bool = True) -> UserInfo:
         """
         Retrieve information on a single user with the given ID or email.
 
         Args:
             id_or_email: A user ID (``usrQBq2RGdihxl3vU``) or email address.
+            collaborations: If ``False``, no collaboration data will be requested
+                from Airtable. This may result in faster responses.
         """
-        return self.users([id_or_email])[0]
+        return self.users([id_or_email], collaborations=collaborations)[0]
 
-    def users(self, ids_or_emails: Iterable[str]) -> List[UserInfo]:
+    def users(
+        self,
+        ids_or_emails: Iterable[str],
+        collaborations: bool = True,
+    ) -> List[UserInfo]:
         """
         Retrieve information on the users with the given IDs or emails.
 
-        Following the Airtable API specification, pyAirtable will perform
-        one API request for each user ID. However, when given a list of emails,
-        pyAirtable only needs to perform one API request for the entire list.
-
-        Read more at `Get user by ID <https://airtable.com/developers/web/api/get-user-by-id>`__
-        and `Get user by email <https://airtable.com/developers/web/api/get-user-by-email>`__.
+        Read more at `Get users by ID or email <https://airtable.com/developers/web/api/get-users-by-id-or-email>`__.
 
         Args:
             ids_or_emails: A sequence of user IDs (``usrQBq2RGdihxl3vU``)
                 or email addresses (or both).
+            collaborations: If ``False``, no collaboration data will be requested
+                from Airtable. This may result in faster responses.
         """
-        users: Dict[str, UserInfo] = {}  # key by user ID to avoid returning duplicates
         user_ids: List[str] = []
         emails: List[str] = []
         for value in ids_or_emails:
-            if "@" in value:
-                emails.append(value)
-            elif is_user_id(value):
-                user_ids.append(value)
-            else:
-                raise ValueError(f"unrecognized user ID or email: {value!r}")
-
-        for user_id in user_ids:
-            response = self.api.request("GET", f"{self.url}/users/{user_id}")
-            info = UserInfo.parse_obj(response)
-            users[info.id] = info
-
-        if emails:
-            params = {"email": emails}
-            response = self.api.request("GET", f"{self.url}/users", params=params)
-            users.update(
-                {
-                    info.id: info
-                    for user_obj in response["users"]
-                    if (info := UserInfo.parse_obj(user_obj))
-                }
-            )
-
+            (emails if "@" in value else user_ids).append(value)
+
+        response = self.api.request(
+            method="GET",
+            url=f"{self.url}/users",
+            params={
+                "id": user_ids,
+                "email": emails,
+                "include": ["collaborations"] if collaborations else [],
+            },
+        )
+        # key by user ID to avoid returning duplicates
+        users = {
+            info.id: info
+            for user_obj in response["users"]
+            if (info := UserInfo.from_api(user_obj, self.api, context=self))
+        }
         return list(users.values())
 
 
diff --git a/pyairtable/models/schema.py b/pyairtable/models/schema.py
index 1023d1c..37f0591 100644
--- a/pyairtable/models/schema.py
+++ b/pyairtable/models/schema.py
@@ -319,38 +319,44 @@ class NestedFieldId(AirtableModel):
     field_id: str
 
 
-class UserInfo(AirtableModel):
+class Collaborations(AirtableModel):
     """
-    Detailed information about a user.
+    The full set of collaborations granted to a user or user group.
 
-    See https://airtable.com/developers/web/api/get-user-by-id
+    See https://airtable.com/developers/web/api/model/collaborations
     """
 
-    id: str
-    name: str
-    email: str
-    state: str
-    is_sso_required: bool
-    is_two_factor_auth_enabled: bool
-    last_activity_time: Optional[str]
-    created_time: Optional[str]
-    enterprise_user_type: Optional[str]
-    invited_to_airtable_by_user_id: Optional[str]
-    is_managed: bool = False
-    collaborations: Optional["Collaborations"]
-    groups: List[NestedId] = pydantic.Field(default_factory=list)
+    base_collaborations: List["Collaborations.BaseCollaboration"] = _FL()
+    interface_collaborations: List["Collaborations.InterfaceCollaboration"] = _FL()
+    workspace_collaborations: List["Collaborations.WorkspaceCollaboration"] = _FL()
 
+    def __bool__(self) -> bool:
+        return bool(
+            self.base_collaborations
+            or self.interface_collaborations
+            or self.workspace_collaborations
+        )
 
-class Collaborations(AirtableModel):
-    """
-    The full set of collaborations granted to a user or user group.
+    @property
+    def bases(self) -> Dict[str, "Collaborations.BaseCollaboration"]:
+        """
+        Mapping of base IDs to collaborations, to make lookups easier.
+        """
+        return {c.base_id: c for c in self.base_collaborations}
 
-    See https://airtable.com/developers/web/api/model/collaborations
-    """
+    @property
+    def interfaces(self) -> Dict[str, "Collaborations.InterfaceCollaboration"]:
+        """
+        Mapping of interface IDs to collaborations, to make lookups easier.
+        """
+        return {c.interface_id: c for c in self.interface_collaborations}
 
-    base_collaborations: List["Collaborations.BaseCollaboration"]
-    interface_collaborations: List["Collaborations.InterfaceCollaboration"]
-    workspace_collaborations: List["Collaborations.WorkspaceCollaboration"]
+    @property
+    def workspaces(self) -> Dict[str, "Collaborations.WorkspaceCollaboration"]:
+        """
+        Mapping of workspace IDs to collaborations, to make lookups easier.
+        """
+        return {c.workspace_id: c for c in self.workspace_collaborations}
 
     class BaseCollaboration(AirtableModel):
         base_id: str
@@ -368,6 +374,28 @@ class Collaborations(AirtableModel):
         permission_level: str
 
 
+class UserInfo(AirtableModel):
+    """
+    Detailed information about a user.
+
+    See https://airtable.com/developers/web/api/get-user-by-id
+    """
+
+    id: str
+    name: str
+    email: str
+    state: str
+    is_sso_required: bool
+    is_two_factor_auth_enabled: bool
+    last_activity_time: Optional[str]
+    created_time: Optional[str]
+    enterprise_user_type: Optional[str]
+    invited_to_airtable_by_user_id: Optional[str]
+    is_managed: bool = False
+    groups: List[NestedId] = _FL()
+    collaborations: "Collaborations" = pydantic.Field(default_factory=Collaborations)
+
+
 class UserGroup(AirtableModel):
     """
     Detailed information about a user group and its members.
@@ -381,7 +409,7 @@ class UserGroup(AirtableModel):
     created_time: str
     updated_time: str
     members: List["UserGroup.Member"]
-    collaborations: Optional["Collaborations"]
+    collaborations: "Collaborations" = pydantic.Field(default_factory=Collaborations)
 
     class Member(AirtableModel):
         user_id: str
diff --git a/tox.ini b/tox.ini
index 426fe49..978a3a9 100644
--- a/tox.ini
+++ b/tox.ini
@@ -23,7 +23,9 @@ deps = -r requirements-dev.txt
 commands = mypy --strict pyairtable tests/test_typing.py
 
 [testenv]
-passenv = AIRTABLE_API_KEY
+passenv =
+    AIRTABLE_API_KEY
+    AIRTABLE_ENTERPRISE_ID
 addopts = -v
 testpaths = tests
 commands = python -m pytest {posargs}


Test patch:
diff --git a/tests/integration/test_integration_enterprise.py b/tests/integration/test_integration_enterprise.py
index 3fef91e..5b1aae6 100644
--- a/tests/integration/test_integration_enterprise.py
+++ b/tests/integration/test_integration_enterprise.py
@@ -1,3 +1,4 @@
+import os
 import uuid
 
 import pytest
@@ -8,6 +9,14 @@ import pyairtable
 pytestmark = [pytest.mark.integration]
 
 
+@pytest.fixture
+def enterprise(api):
+    try:
+        return api.enterprise(os.environ["AIRTABLE_ENTERPRISE_ID"])
+    except KeyError:
+        pytest.skip("test requires AIRTABLE_ENTERPRISE_ID")
+
+
 @pytest.fixture
 def workspace_id():
     return "wsp0HnyXmNnKzc5ng"
@@ -38,6 +47,31 @@ def blank_base(workspace: pyairtable.Workspace):
         base.delete()
 
 
+def test_user(enterprise: pyairtable.Enterprise):
+    """
+    Test that we can retrieve information about the current logged-in user.
+    """
+    user_id = enterprise.api.whoami()["id"]
+    assert user_id == enterprise.user(user_id).id
+
+
+def test_user__invalid(enterprise):
+    with pytest.raises(HTTPError):
+        enterprise.user("invalidUserId")
+
+
+def test_users(enterprise: pyairtable.Enterprise):
+    """
+    Test that we can retrieve information about an enterprise
+    and retrieve user information by ID or by email.
+    """
+    user_ids = enterprise.info().user_ids[:5]
+    users_from_ids = enterprise.users(user_ids)
+    assert {u.id for u in users_from_ids} == set(user_ids)
+    users_from_emails = enterprise.users(u.email for u in users_from_ids)
+    assert {u.id for u in users_from_emails} == set(user_ids)
+
+
 def test_create_table(blank_base: pyairtable.Base):
     """
     Test that we can create a new table on an existing base.
diff --git a/tests/test_api_enterprise.py b/tests/test_api_enterprise.py
index e62b065..d74eaf3 100644
--- a/tests/test_api_enterprise.py
+++ b/tests/test_api_enterprise.py
@@ -13,25 +13,17 @@ def enterprise(api):
 
 @pytest.fixture
 def enterprise_mocks(enterprise, requests_mock, sample_json):
-    user_json = sample_json("UserInfo")
-    group_json = sample_json("UserGroup")
     m = Mock()
-    m.user_id = user_json["id"]
-    m.get_info = requests_mock.get(
-        enterprise.url,
-        json=sample_json("EnterpriseInfo"),
-    )
-    m.get_user = requests_mock.get(
-        f"{enterprise.url}/users/{m.user_id}",
-        json=user_json,
-    )
-    m.get_users = requests_mock.get(
-        f"{enterprise.url}/users",
-        json={"users": [user_json]},
-    )
+    m.json_user = sample_json("UserInfo")
+    m.json_users = {"users": [m.json_user]}
+    m.json_group = sample_json("UserGroup")
+    m.user_id = m.json_user["id"]
+    m.group_id = m.json_group["id"]
+    m.get_info = requests_mock.get(enterprise.url, json=sample_json("EnterpriseInfo"))
+    m.get_users = requests_mock.get(f"{enterprise.url}/users", json=m.json_users)
     m.get_group = requests_mock.get(
-        enterprise.api.build_url(f"meta/groups/{group_json['id']}"),
-        json=group_json,
+        enterprise.api.build_url(f"meta/groups/{m.json_group['id']}"),
+        json=m.json_group,
     )
     return m
 
@@ -50,7 +42,24 @@ def test_info(enterprise, enterprise_mocks):
 def test_user(enterprise, enterprise_mocks):
     user = enterprise.user(enterprise_mocks.user_id)
     assert isinstance(user, UserInfo)
-    assert enterprise_mocks.get_user.call_count == 1
+    assert enterprise_mocks.get_users.call_count == 1
+    assert user.collaborations
+    assert "appLkNDICXNqxSDhG" in user.collaborations.bases
+    assert "pbdyGA3PsOziEHPDE" in user.collaborations.interfaces
+    assert "wspmhESAta6clCCwF" in user.collaborations.workspaces
+
+
+def test_user__no_collaboration(enterprise, enterprise_mocks):
+    del enterprise_mocks.json_users["users"][0]["collaborations"]
+
+    user = enterprise.user(enterprise_mocks.user_id, collaborations=False)
+    assert isinstance(user, UserInfo)
+    assert enterprise_mocks.get_users.call_count == 1
+    assert not enterprise_mocks.get_users.last_request.qs.get("include")
+    assert not user.collaborations  # test for Collaborations.__bool__
+    assert not user.collaborations.bases
+    assert not user.collaborations.interfaces
+    assert not user.collaborations.workspaces
 
 
 @pytest.mark.parametrize(
@@ -69,15 +78,27 @@ def test_users(enterprise, enterprise_mocks, search_for):
     assert user.state == "provisioned"
 
 
-def test_users__invalid_value(enterprise, enterprise_mocks):
-    with pytest.raises(ValueError):
-        enterprise.users(["not an ID or email"])
+def test_group(enterprise, enterprise_mocks):
+    grp = enterprise.group("ugp1mKGb3KXUyQfOZ")
+    assert enterprise_mocks.get_group.call_count == 1
+    assert isinstance(grp, UserGroup)
+    assert grp.id == "ugp1mKGb3KXUyQfOZ"
+    assert grp.name == "Group name"
+    assert grp.members[0].email == "foo@bar.com"
+    assert grp.collaborations
+    assert "appLkNDICXNqxSDhG" in grp.collaborations.bases
+    assert "pbdyGA3PsOziEHPDE" in grp.collaborations.interfaces
+    assert "wspmhESAta6clCCwF" in grp.collaborations.workspaces
 
 
-def test_group(enterprise, enterprise_mocks):
-    info = enterprise.group("ugp1mKGb3KXUyQfOZ")
+def test_group__no_collaboration(enterprise, enterprise_mocks):
+    del enterprise_mocks.json_group["collaborations"]
+
+    grp = enterprise.group(enterprise_mocks.group_id, collaborations=False)
+    assert isinstance(grp, UserGroup)
     assert enterprise_mocks.get_group.call_count == 1
-    assert isinstance(info, UserGroup)
-    assert info.id == "ugp1mKGb3KXUyQfOZ"
-    assert info.name == "Group name"
-    assert info.members[0].email == "foo@bar.com"
+    assert not enterprise_mocks.get_group.last_request.qs.get("include")
+    assert not grp.collaborations  # test for Collaborations.__bool__
+    assert not grp.collaborations.bases
+    assert not grp.collaborations.interfaces
+    assert not grp.collaborations.workspaces