Add get_userinfo_by_id method to ModuleApi

changelog.d/9581.feature

@@ -0,0 +1 @@
+Add get_user_by_id method to ModuleApi.

synapse/module_api/__init__.py

@@ -45,7 +45,7 @@
 from synapse.storage.database import DatabasePool, LoggingTransaction
 from synapse.storage.databases.main.roommember import ProfileInfo
 from synapse.storage.state import StateFilter
-from synapse.types import JsonDict, Requester, UserID, create_requester
+from synapse.types import JsonDict, Requester, User, UserID, create_requester
 from synapse.util import Clock
 from synapse.util.caches.descriptors import cached
 
@@ -174,6 +174,34 @@ def email_app_name(self) -> str:
         """The application name configured in the homeserver's configuration."""
         return self._hs.config.email.email_app_name
 
+    async def get_user_by_id(self, user_id: str) -> Optional[User]:
+        """Get User by user_id
+
+        Args:
+            user_id: Fully qualified user id.
+        Returns:
+            User object if a user was found, otherwise None
+        """
+        user_data = await self._store.get_user_by_id(user_id)
+        if user_data:
+            app_service = None
+            if user_data.get("appservice_id"):
+                app_service = self._store.get_app_service_by_id(
+                    user_data.get("appservice_id")
+                )
+            return User(
+                user=UserID.from_string(user_data.get("name")),
+                creation_ts=user_data.get("creation_ts"),
+                consent_version=user_data.get("consent_version"),
+                consent_server_notice_sent=user_data.get("consent_server_notice_sent"),
+                app_service=app_service,
+                shadow_banned=user_data.get("shadow_banned"),
+                deactivated=user_data.get("deactivated"),
+                is_guest=user_data.get("is_guest"),
+                admin=user_data.get("admin"),
+                user_type=user_data.get("user_type"),
+            )
+
     async def get_user_by_req(
         self,
         req: SynapseRequest,

synapse/types.py

@@ -751,3 +751,32 @@ def get_verify_key_from_cross_signing_key(key_info):
     # and return that one key
     for key_id, key_data in keys.items():
         return (key_id, decode_verify_key_bytes(key_id, decode_base64(key_data)))
+
+
+@attr.s(frozen=True, slots=True)
+class User:
+    """Represents a user object.
+
+    Attributes:
+        user:  ID of the user.
+        admin:  True if the user is an admin.
+        app_service:  Application service that created this user.
+        consent_server_notice_sent:  Version of policy documents the user has been sent.
+        consent_version:  Version of policy documents the user has consented to.
+        creation_ts:  Creation timestamp of the user.
+        deactivated:  True if the user has been deactivated.
+        is_guest:  True if the user is a guest user.
+        shadow_banned:  True if the user has been shadow-banned.
+        user_type:  User type (None for normal user, 'support' and 'bot' other options).
+    """
+
+    user = attr.ib(type="UserID")
+    admin = attr.ib(type=bool, default=False, converter=bool)
+    app_service = attr.ib(type=Optional["ApplicationService"], default=None)
+    consent_server_notice_sent = attr.ib(type=Optional[str], default=None)
+    consent_version = attr.ib(type=Optional[str], default=None)
+    creation_ts = attr.ib(type=int, default=0)
+    deactivated = attr.ib(type=bool, default=False, converter=bool)
+    is_guest = attr.ib(type=bool, default=False, converter=bool)
+    shadow_banned = attr.ib(type=bool, default=False, converter=bool)
+    user_type = attr.ib(type=str, default=None)

tests/module_api/test_api.py

@@ -79,6 +79,16 @@ def test_can_register_user(self):
         displayname = self.get_success(self.store.get_profile_displayname("bob"))
         self.assertEqual(displayname, "Bobberino")
 
+    def test_get_user_by_id(self):
+        user_id = self.register_user("alice", "1234")
+        found_user = self.get_success(self.module_api.get_user_by_id(user_id))
+        self.assertEqual(found_user.user.to_string(), user_id)
+        self.assertIdentical(found_user.admin, False)
+
+    def test_get_user_by_id__no_user_found(self):
+        found_user = self.get_success(self.module_api.get_user_by_id("@alice:test"))
+        self.assertIsNone(found_user)
+
     def test_sending_events_into_room(self):
         """Tests that a module can send events into a room"""
         # Mock out create_and_send_nonmember_event to check whether events are being sent

tests/test_types.py

@@ -13,7 +13,14 @@
 # limitations under the License.
 
 from synapse.api.errors import SynapseError
-from synapse.types import GroupID, RoomAlias, UserID, map_username_to_mxid_localpart
+from synapse.appservice import ApplicationService
+from synapse.types import (
+    GroupID,
+    RoomAlias,
+    User,
+    UserID,
+    map_username_to_mxid_localpart,
+)
 
 from tests import unittest
 
@@ -104,3 +111,49 @@ def testNonAscii(self):
         # this should work with either a unicode or a bytes
         self.assertEqual(map_username_to_mxid_localpart("têst"), "t=c3=aast")
         self.assertEqual(map_username_to_mxid_localpart("têst".encode()), "t=c3=aast")
+
+
+class UserTestCase(unittest.TestCase):
+    def test_build(self):
+        user = User(user=UserID.from_string("@alice:test"),)
+        self.assertEqual(user.user.to_string(), "@alice:test")
+        self.assertEqual(user.creation_ts, 0)
+        self.assertIsNone(user.consent_version)
+        self.assertIsNone(user.consent_server_notice_sent)
+        self.assertIsNone(user.app_service)
+        self.assertIdentical(user.shadow_banned, False)
+        self.assertIdentical(user.deactivated, False)
+        self.assertIdentical(user.is_guest, False)
+        self.assertIdentical(user.admin, False)
+        self.assertIsNone(user.user_type)
+
+    def test_build__all_values(self):
+        appservice = ApplicationService(
+            None,
+            "test",
+            id="foo",
+            rate_limited=True,
+            sender="@as:test",
+        )
+        user = User(
+            user=UserID.from_string("@alice:test"),
+            creation_ts=1234567890,
+            consent_version="1.0",
+            consent_server_notice_sent="2.0",
+            app_service=appservice,
+            shadow_banned=True,
+            deactivated=True,
+            is_guest=True,
+            admin=True,
+            user_type="support",
+        )
+        self.assertEqual(user.user.to_string(), "@alice:test")
+        self.assertEqual(user.creation_ts, 1234567890)
+        self.assertEqual(user.consent_version, "1.0")
+        self.assertEqual(user.consent_server_notice_sent, "2.0")
+        self.assertEqual(user.app_service, appservice)
+        self.assertIdentical(user.shadow_banned, True)
+        self.assertIdentical(user.deactivated, True)
+        self.assertIdentical(user.is_guest, True)
+        self.assertIdentical(user.admin, True)
+        self.assertEqual(user.user_type, "support")