typing: Add UserModel dataclass

master/buildbot/db/users.py

@@ -14,22 +14,64 @@
 # Copyright Buildbot Team Members
 
 
+from __future__ import annotations
+
+import dataclasses
+from typing import TYPE_CHECKING
+
 import sqlalchemy as sa
+from twisted.python import deprecate
+from twisted.python import versions
 
 from buildbot.db import base
 from buildbot.util import identifiers
+from buildbot.warnings import warn_deprecated
+
+if TYPE_CHECKING:
+    from twisted.internet import defer
+
+
+@dataclasses.dataclass
+class UserModel:
+    uid: int
+    identifier: str
+    bb_username: str | None = None
+    bb_password: str | None = None
+    attributes: dict[str, str] | None = None
 
+    # For backward compatibility
+    def __getitem__(self, key: str):
+        warn_deprecated(
+            '4.1.0',
+            (
+                'UsersConnectorComponent '
+                'getUser, getUserByUsername, and getUsers '
+                'no longer return User as dictionnaries. '
+                'Usage of [] accessor is deprecated: please access the member directly'
+            ),
+        )
 
-class UsDict(dict):
+        if hasattr(self, key):
+            return getattr(self, key)
+
+        if self.attributes is not None and key in self.attributes:
+            return self.attributes[key]
+
+        raise KeyError(key)
+
+
+@deprecate.deprecated(versions.Version("buildbot", 4, 1, 0), UserModel)
+class UsDict(UserModel):
     pass
 
 
 class UsersConnectorComponent(base.DBConnectorComponent):
-    # returns a Deferred that returns a value
-    def findUserByAttr(self, identifier, attr_type, attr_data, _race_hook=None):
+    def findUserByAttr(
+        self, identifier: str, attr_type: str, attr_data: str, _race_hook=None
+    ) -> defer.Deferred[int]:
         # note that since this involves two tables, self.findSomethingId is not
         # helpful
-        def thd(conn, no_recurse=False, identifier=identifier):
+        def thd(conn, no_recurse=False, identifier=identifier) -> int:
             tbl = self.db.model.users
             tbl_info = self.db.model.users_info
 
@@ -86,10 +128,9 @@ def thd(conn, no_recurse=False, identifier=identifier):
 
         return self.db.pool.do(thd)
 
-    # returns a Deferred that returns a value
     @base.cached("usdicts")
-    def getUser(self, uid):
-        def thd(conn):
+    def getUser(self, uid: int) -> defer.Deferred[UserModel | None]:
+        def thd(conn) -> UserModel | None:
             tbl = self.db.model.users
             tbl_info = self.db.model.users_info
 
@@ -103,28 +144,25 @@ def thd(conn):
             q = tbl_info.select().where(tbl_info.c.uid == uid)
             rows = conn.execute(q).fetchall()
 
-            return self.thd_createUsDict(users_row, rows)
+            return self._model_from_row(users_row, rows)
 
         return self.db.pool.do(thd)
 
-    def thd_createUsDict(self, users_row, rows):
-        # make UsDict to return
-        usdict = UsDict()
-        for row in rows:
-            usdict[row.attr_type] = row.attr_data
-
-        # add the users_row data *after* the attributes in case attr_type
-        # matches one of these keys.
-        usdict['uid'] = users_row.uid
-        usdict['identifier'] = users_row.identifier
-        usdict['bb_username'] = users_row.bb_username
-        usdict['bb_password'] = users_row.bb_password
-
-        return usdict
+    def _model_from_row(self, users_row, attribute_rows=None):
+        attributes = None
+        if attribute_rows is not None:
+            attributes = {row.attr_type: row.attr_data for row in attribute_rows}
+        return UserModel(
+            uid=users_row.uid,
+            identifier=users_row.identifier,
+            bb_username=users_row.bb_username,
+            bb_password=users_row.bb_password,
+            attributes=attributes,
+        )
 
     # returns a Deferred that returns a value
-    def getUserByUsername(self, username):
-        def thd(conn):
+    def getUserByUsername(self, username: str | None) -> defer.Deferred[UserModel | None]:
+        def thd(conn) -> UserModel | None:
             tbl = self.db.model.users
             tbl_info = self.db.model.users_info
 
@@ -138,34 +176,28 @@ def thd(conn):
             q = tbl_info.select().where(tbl_info.c.uid == users_row.uid)
             rows = conn.execute(q).fetchall()
 
-            return self.thd_createUsDict(users_row, rows)
+            return self._model_from_row(users_row, rows)
 
         return self.db.pool.do(thd)
 
-    # returns a Deferred that returns a value
-    def getUsers(self):
-        def thd(conn):
+    def getUsers(self) -> defer.Deferred[list[UserModel]]:
+        def thd(conn) -> list[UserModel]:
             tbl = self.db.model.users
             rows = conn.execute(tbl.select()).fetchall()
 
-            dicts = []
-            if rows:
-                for row in rows:
-                    ud = {"uid": row.uid, "identifier": row.identifier}
-                    dicts.append(ud)
-            return dicts
+            return [self._model_from_row(row, attribute_rows=None) for row in rows]
 
         return self.db.pool.do(thd)
 
     # returns a Deferred that returns None
     def updateUser(
         self,
-        uid=None,
-        identifier=None,
-        bb_username=None,
-        bb_password=None,
-        attr_type=None,
-        attr_data=None,
+        uid: int | None = None,
+        identifier: str | None = None,
+        bb_username: str | None = None,
+        bb_password: str | None = None,
+        attr_type: str | None = None,
+        attr_data: str | None = None,
         _race_hook=None,
     ):
         def thd(conn):
@@ -240,8 +272,8 @@ def thd(conn):
         return self.db.pool.do(thd)
 
     # returns a Deferred that returns a value
-    def identifierToUid(self, identifier):
-        def thd(conn):
+    def identifierToUid(self, identifier) -> defer.Deferred[int | None]:
+        def thd(conn) -> int | None:
             tbl = self.db.model.users
 
             q = tbl.select().where(tbl.c.identifier == identifier)

master/buildbot/process/users/manual.py

@@ -73,10 +73,17 @@ def formatResults(self, op, results):
             formatted_results += "user(s) found:\n"
             for user in results:
                 if user:
-                    for key in sorted(user.keys()):
-                        if key != 'bb_password':
-                            formatted_results += f"{key}: {user[key]}\n"
-                    formatted_results += "\n"
+                    formatted_results += (
+                        f"uid: {user.uid}\n"
+                        f"identifier: {user.identifier}\n"
+                        f"bb_username: {user.bb_username}\n"
+                    )
+                    if user.attributes:
+                        formatted_results += "attributes:\n"
+                        formatted_results += (
+                            ''.join(f"\t{key}: {value}\n" for key, value in user.attributes.items())
+                            + '\n'
+                        )
                 else:
                     formatted_results += "no match found\n"
         return formatted_results

master/buildbot/process/users/users.py

@@ -13,17 +13,22 @@
 #
 # Copyright Buildbot Team Members
 
+from __future__ import annotations
 
 import os
 from binascii import hexlify
 from hashlib import sha1
+from typing import TYPE_CHECKING
 
 from twisted.internet import defer
 from twisted.python import log
 
 from buildbot.util import bytes2unicode
 from buildbot.util import unicode2bytes
 
+if TYPE_CHECKING:
+    from buildbot.db.users import UserModel
+
 # TODO: fossil comes from a plugin. We should have an API that plugins could use to
 # register allowed user types.
 srcs = ['git', 'svn', 'hg', 'cvs', 'darcs', 'bzr', 'fossil']
@@ -63,10 +68,10 @@ def createUserObject(master, author, src=None):
     )
 
 
-def _extractContact(usdict, contact_types, uid):
-    if usdict:
+def _extractContact(user: UserModel | None, contact_types, uid):
+    if user is not None and user.attributes is not None:
         for type in contact_types:
-            contact = usdict.get(type)
+            contact = user.attributes.get(type)
             if contact:
                 break
     else:

master/buildbot/test/fakedb/users.py

@@ -13,9 +13,11 @@
 #
 # Copyright Buildbot Team Members
 
+from __future__ import annotations
 
 from twisted.internet import defer
 
+from buildbot.db.users import UserModel
 from buildbot.test.fakedb.base import FakeDBComponent
 from buildbot.test.fakedb.row import Row
 
@@ -68,16 +70,25 @@ def insert_test_data(self, rows):
                         "attr_data": row.attr_data,
                     })
 
-    def _user2dict(self, uid):
-        usdict = None
+    def _model_from_uid(self, uid: int) -> UserModel | None:
+        model = None
         if uid in self.users:
             usdict = self.users[uid]
+            model = UserModel(
+                uid=uid,
+                identifier=usdict['identifier'],
+                bb_username=usdict.get('bb_username'),
+                bb_password=usdict.get('bb_password'),
+                attributes=None,
+            )
             if uid in self.users_info:
                 infos = self.users_info[uid]
+                attributes = {}
                 for attr in infos:
-                    usdict[attr['attr_type']] = attr['attr_data']
-            usdict['uid'] = uid
-        return usdict
+                    attributes[attr['attr_type']] = attr['attr_data']
+                model.attributes = attributes
+
+        return model
 
     def nextId(self):
         self.id_num += 1
@@ -96,17 +107,17 @@ def findUserByAttr(self, identifier, attr_type, attr_data):
         self.db.insert_test_data([UserInfo(uid=uid, attr_type=attr_type, attr_data=attr_data)])
         return defer.succeed(uid)
 
-    def getUser(self, uid):
+    def getUser(self, uid) -> defer.Deferred[UserModel | None]:
         usdict = None
         if uid in self.users:
-            usdict = self._user2dict(uid)
+            usdict = self._model_from_uid(uid)
         return defer.succeed(usdict)
 
-    def getUserByUsername(self, username):
+    def getUserByUsername(self, username) -> defer.Deferred[UserModel | None]:
         usdict = None
         for uid, user in self.users.items():
             if user['bb_username'] == username:
-                usdict = self._user2dict(uid)
+                usdict = self._model_from_uid(uid)
         return defer.succeed(usdict)
 
     def updateUser(