Merge pull request #1965 from dudanogueira/expose-include-last-user-time-and-created-at

expose include_last_used_time parameter, return created_at and api_key_first_letters

Author: dirkkul

integration/test_users.py

@@ -214,3 +214,82 @@ def test_list_all_users(client_factory: ClientFactory) -> None:
         assert len(dynamic_users) == 5
         for i in range(5):
             client.users.db.delete(user_id=f"list-all-{i}")
+
+
+def test_get_user_created_at_and_api_key_first_letters(client_factory: ClientFactory) -> None:
+    with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client:
+        if client._connection._weaviate_version.is_lower_than(1, 30, 0):
+            pytest.skip("This test requires Weaviate 1.30.0 or higher")
+
+        randomUserName = "new-user" + str(random.randint(1, 1000))
+        client.users.db.create(user_id=randomUserName)
+        try:
+            user = client.users.db.get(user_id=randomUserName)
+            assert user is not None
+            assert user.created_at is not None
+            assert user.api_key_first_letters is not None
+            assert len(user.api_key_first_letters) > 0
+        finally:
+            client.users.db.delete(user_id=randomUserName)
+
+
+def test_get_user_include_last_used_time(client_factory: ClientFactory) -> None:
+    with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client:
+        if client._connection._weaviate_version.is_lower_than(1, 30, 0):
+            pytest.skip("This test requires Weaviate 1.30.0 or higher")
+
+        randomUserName = "new-user" + str(random.randint(1, 1000))
+        apiKey = client.users.db.create(user_id=randomUserName)
+        try:
+            # log in with the new user to generate a lastUsedAt timestamp
+            with weaviate.connect_to_local(
+                port=RBAC_PORTS[0],
+                grpc_port=RBAC_PORTS[1],
+                auth_credentials=Auth.api_key(apiKey),
+            ) as client2:
+                assert client2.users.get_my_user().user_id == randomUserName
+
+            # without include_last_used_time, last_used_time should be None
+            user = client.users.db.get(user_id=randomUserName)
+            assert user is not None
+            assert user.last_used_time is None
+
+            # with include_last_used_time=True, last_used_time should be populated
+            user = client.users.db.get(user_id=randomUserName, include_last_used_time=True)
+            assert user is not None
+            assert user.last_used_time is not None
+        finally:
+            client.users.db.delete(user_id=randomUserName)
+
+
+def test_list_all_include_last_used_time(client_factory: ClientFactory) -> None:
+    with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client:
+        if client._connection._weaviate_version.is_lower_than(1, 30, 0):
+            pytest.skip("This test requires Weaviate 1.30.0 or higher")
+
+        randomUserName = "new-user" + str(random.randint(1, 1000))
+        apiKey = client.users.db.create(user_id=randomUserName)
+        try:
+            # log in with the new user to generate a lastUsedAt timestamp
+            with weaviate.connect_to_local(
+                port=RBAC_PORTS[0],
+                grpc_port=RBAC_PORTS[1],
+                auth_credentials=Auth.api_key(apiKey),
+            ) as client2:
+                assert client2.users.get_my_user().user_id == randomUserName
+
+            # without include_last_used_time, last_used_time should be None
+            users = client.users.db.list_all()
+            target = next((u for u in users if u.user_id == randomUserName), None)
+            assert target is not None
+            assert target.created_at is not None
+            assert target.api_key_first_letters is not None
+            assert target.last_used_time is None
+
+            # with include_last_used_time=True, last_used_time should be populated
+            users = client.users.db.list_all(include_last_used_time=True)
+            target = next((u for u in users if u.user_id == randomUserName), None)
+            assert target is not None
+            assert target.last_used_time is not None
+        finally:
+            client.users.db.delete(user_id=randomUserName)

weaviate/rbac/models.py

@@ -140,6 +140,9 @@ class WeaviateDBUserRoleNames(TypedDict):
     groups: List[str]
     active: bool
     dbUserType: str
+    createdAt: NotRequired[str]
+    lastUsedAt: NotRequired[str]
+    apiKeyFirstLetters: NotRequired[str]
 
 
 class _Action:

weaviate/users/async_.pyi

@@ -57,8 +57,10 @@ class _UsersDBAsync(_UsersDBExecutor[ConnectionAsync]):
     async def rotate_key(self, *, user_id: str) -> str: ...
     async def activate(self, *, user_id: str) -> bool: ...
     async def deactivate(self, *, user_id: str, revoke_key: bool = False) -> bool: ...
-    async def get(self, *, user_id: str) -> Optional[UserDB]: ...
-    async def list_all(self) -> List[UserDB]: ...
+    async def get(
+        self, *, user_id: str, include_last_used_time: bool = False
+    ) -> Optional[UserDB]: ...
+    async def list_all(self, *, include_last_used_time: bool = False) -> List[UserDB]: ...
 
 class _UsersAsync(_UsersExecutor[ConnectionAsync]):
     async def get_my_user(self) -> OwnUser: ...

weaviate/users/base.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from typing import Any, Dict, Generic, List, Literal, Optional, Union, cast, overload
 
 from httpx import Response
@@ -20,6 +21,16 @@
 )
 from weaviate.util import _decode_json_response_dict, escape_string
 
+# Go's zero time value, returned by the server when a timestamp is not set
+_GO_ZERO_TIME = datetime(1, 1, 1, 0, 0, tzinfo=timezone.utc)
+
+
+def _parse_last_used_at(value: Optional[str]) -> Optional[datetime]:
+    if value is None:
+        return None
+    dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    return None if dt == _GO_ZERO_TIME else dt
+
 
 class _BaseExecutor(Generic[ConnectionType]):
     def __init__(self, connection: ConnectionType):
@@ -451,35 +462,51 @@ def resp(res: Response) -> bool:
             status_codes=_ExpectedStatusCodes(ok_in=[200, 409], error="Deactivate user"),
         )
 
-    def get(self, *, user_id: str) -> executor.Result[Optional[UserDB]]:
+    def get(
+        self, *, user_id: str, include_last_used_time: bool = False
+    ) -> executor.Result[Optional[UserDB]]:
         """Get all information about an user.
 
         Args:
             user_id: The id of the user.
+            include_last_used_time: If True, the last used time of the user will be included.
         """
 
         def resp(res: Response) -> Optional[UserDB]:
             if res.status_code == 404:
                 return None
             parsed = _decode_json_response_dict(res, "Get user")
             assert parsed is not None
+            created_at = parsed.get("createdAt")
             return UserDB(
                 user_id=parsed["userId"],
                 role_names=parsed["roles"],
                 active=parsed["active"],
                 user_type=UserTypes(parsed["dbUserType"]),
+                created_at=(
+                    datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+                    if created_at is not None
+                    else None
+                ),
+                last_used_time=_parse_last_used_at(parsed.get("lastUsedAt")),
+                api_key_first_letters=parsed.get("apiKeyFirstLetters"),
             )
 
         return executor.execute(
             response_callback=resp,
             method=self._connection.get,
             path=f"/users/db/{user_id}",
+            params={"includeLastUsedTime": include_last_used_time},
             error_msg=f"Could not get user '{user_id}'",
             status_codes=_ExpectedStatusCodes(ok_in=[200, 404], error="get user"),
         )
 
-    def list_all(self) -> executor.Result[List[UserDB]]:
-        """List all DB users."""
+    def list_all(self, *, include_last_used_time: bool = False) -> executor.Result[List[UserDB]]:
+        """List all DB users.
+
+        Args:
+            include_last_used_time: If True, the last used time of each user will be included.
+        """
 
         def resp(res: Response) -> List[UserDB]:
             parsed = _decode_json_response_dict(res, "Get user")
@@ -490,6 +517,13 @@ def resp(res: Response) -> List[UserDB]:
                     role_names=user["roles"],
                     active=user["active"],
                     user_type=UserTypes(user["dbUserType"]),
+                    created_at=(
+                        datetime.fromisoformat(ca.replace("Z", "+00:00"))
+                        if (ca := user.get("createdAt")) is not None
+                        else None
+                    ),
+                    last_used_time=_parse_last_used_at(user.get("lastUsedAt")),
+                    api_key_first_letters=user.get("apiKeyFirstLetters"),
                 )
                 for user in cast(List[WeaviateDBUserRoleNames], parsed)
             ]
@@ -498,6 +532,7 @@ def resp(res: Response) -> List[UserDB]:
             response_callback=resp,
             method=self._connection.get,
             path="/users/db",
+            params={"includeLastUsedTime": include_last_used_time},
             error_msg="Could not list all users",
             status_codes=_ExpectedStatusCodes(ok_in=[200], error="list all users"),
         )

weaviate/users/sync.pyi

@@ -57,8 +57,8 @@ class _UsersDB(_UsersDBExecutor[ConnectionSync]):
     def rotate_key(self, *, user_id: str) -> str: ...
     def activate(self, *, user_id: str) -> bool: ...
     def deactivate(self, *, user_id: str, revoke_key: bool = False) -> bool: ...
-    def get(self, *, user_id: str) -> Optional[UserDB]: ...
-    def list_all(self) -> List[UserDB]: ...
+    def get(self, *, user_id: str, include_last_used_time: bool = False) -> Optional[UserDB]: ...
+    def list_all(self, *, include_last_used_time: bool = False) -> List[UserDB]: ...
 
 class _Users(_UsersExecutor[ConnectionSync]):
     def get_my_user(self) -> OwnUser: ...

weaviate/users/users.py

@@ -1,5 +1,6 @@
-from dataclasses import dataclass
-from typing import Dict, Final, List, Literal
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Dict, Final, List, Literal, Optional
 
 from weaviate.rbac.models import (
     Role,
@@ -29,6 +30,9 @@ class UserBase:
 class UserDB(UserBase):
     user_type: UserTypes
     active: bool
+    created_at: Optional[datetime] = field(default=None)
+    last_used_time: Optional[datetime] = field(default=None)
+    api_key_first_letters: Optional[str] = field(default=None)
 
 
 @dataclass