[APP-5754] Python GetFragmentHistory SDK changes

src/viam/app/app_client.py

@@ -8,10 +8,9 @@
 
 from viam import logging
 from viam.app._logs import _LogsStream, _LogsStreamWithIterator
+from viam.proto.app import AddRoleRequest, APIKeyWithAuthorizations, AppServiceStub
 from viam.proto.app import (
-    AddRoleRequest,
-    APIKeyWithAuthorizations,
-    AppServiceStub,
+    AuthenticatorInfo,
     Authorization,
     AuthorizedPermissions,
     ChangeRoleRequest,
@@ -49,8 +48,11 @@
     DeleteRobotRequest,
 )
 from viam.proto.app import Fragment as FragmentPB
+from viam.proto.app import FragmentHistoryEntry as FragmentHistoryEntryPB
 from viam.proto.app import FragmentVisibility as FragmentVisibilityPB
 from viam.proto.app import (
+    GetFragmentHistoryRequest,
+    GetFragmentHistoryResponse,
     GetFragmentRequest,
     GetFragmentResponse,
     GetLocationRequest,
@@ -383,6 +385,120 @@ def proto(self) -> FragmentPB:
         )
 
 
+# class AuthenticatorInfo:
+#     class AuthenticationType(str, Enum):
+#         """
+#         AuthenticationType specifies the authentication method used by the caller in editing the fragment.
+#         """
+#
+#         WEB_OAUTH = "web_oauth"
+#         """
+#         Caller authenticated via oauth, presumably via Viam app frontend.
+#         """
+#
+#         API_KEY = "api_key"
+#         """
+#         Caller authenticated via api key.
+#         """
+#
+#         ROBOT_PART_SECRET = "robot_part_secret"
+#         """
+#         Caller authenticated via robot part secret.
+#         """
+#
+#         LOCATION_SECRET = "location_secret"
+#         """
+#         Caller authenticated via location secret.
+#         """
+#
+#         UNSPECIFIED = "unspecified"
+#
+#         """
+#         Unspecified authentication type.
+#         """
+#
+#         @classmethod
+#         def from_proto(cls, authentication_type: AuthenticationTypePB.ValueType):
+#             if authentication_type == AuthenticationTypePB.AUTHENTICATION_TYPE_WEB_OAUTH:
+#                 return AuthenticatorInfo.AuthenticationType.WEB_OAUTH
+#             if authentication_type == AuthenticationTypePB.AUTHENTICATION_TYPE_API_KEY:
+#                 return AuthenticatorInfo.AuthenticationType.API_KEY
+#             if authentication_type == AuthenticationTypePB.AUTHENTICATION_TYPE_LOCATION_SECRET:
+#                 return AuthenticatorInfo.AuthenticationType.LOCATION_SECRET
+#             if authentication_type == AuthenticationTypePB.AUTHENTICATION_TYPE_ROBOT_PART_SECRET:
+#                 return AuthenticatorInfo.AuthenticationType.ROBOT_PART_SECRET
+#             return AuthenticatorInfo.AuthenticationType.UNSPECIFIED
+#
+#         def to_proto(self) -> AuthenticationTypePB.ValueType:
+#             if self == self.WEB_OAUTH:
+#                 return AuthenticationTypePB.AUTHENTICATION_TYPE_WEB_OAUTH
+#             if self == self.API_KEY:
+#                 return AuthenticationTypePB.AUTHENTICATION_TYPE_API_KEY
+#             if self == self.LOCATION_SECRET:
+#                 return AuthenticationTypePB.AUTHENTICATION_TYPE_LOCATION_SECRET
+#             if self == self.ROBOT_PART_SECRET:
+#                 return AuthenticationTypePB.AUTHENTICATION_TYPE_ROBOT_PART_SECRET
+#             return AuthenticationTypePB.AUTHENTICATION_TYPE_UNSPECIFIED
+#
+#     @classmethod
+#     def from_proto(cls, auth_info: AuthenticatorInfoPB) -> Self:
+#         self = cls()
+#         self.type = AuthenticatorInfo.AuthenticationType.from_proto(auth_info.type)
+#         self.value = auth_info.value
+#         self.is_deactivated = auth_info.is_deactivated
+#         return self
+#
+#     @classmethod
+#     def to_proto(self) -> AuthenticatorInfoPB:
+#         return AuthenticatorInfoPB(
+#             is_deactivated=self.is_deactivated,
+#             value=self.value,
+#             type=AuthenticatorInfo.AuthenticationType.to_proto(self.type)
+#         )
+#
+#     type: AuthenticationType
+#     value: str
+#     is_deactivated: bool
+
+
+class FragmentHistoryEntry:
+    """A class that mirror the `FragmentHistoryEntry` proto message.
+
+    Use this class to make the attributes of a `viam.proto.app.FragmentHistoryEntry` more accessible and easier to read/interpret.
+    """
+
+    @classmethod
+    def from_proto(cls, fragment_history_entry: FragmentHistoryEntryPB) -> Self:
+        """Create a `FragmentHistoryEntry` from the .proto defined `FragmentHistoryEntry`.
+
+        Args:
+            fragment_history_entry (viam.proto.app.FragmentHistoryEntry): The object to copy from.
+
+        Returns:
+            FragmentHistoryEntry: The `FragmentHistoryEntry`.
+        """
+        self = cls()
+        self.fragment = fragment_history_entry.fragment
+        self.edited_on = fragment_history_entry.edited_on.ToDatetime()
+        self.old = Fragment.from_proto(fragment_history_entry.old)
+        self.edited_by = fragment_history_entry.edited_by
+        return self
+
+    fragment: str
+    edited_on: datetime
+    old: Fragment
+    edited_by: AuthenticatorInfo
+
+    @property
+    def proto(self) -> FragmentHistoryEntryPB:
+        return FragmentHistoryEntryPB(
+            fragment=self.fragment,
+            edited_on=datetime_to_timestamp(self.edited_on),
+            edited_by=self.edited_by,
+            old=self.old.proto if self.old else None,
+        )
+
+
 class RobotPartHistoryEntry:
     """A class that mirrors the `RobotPartHistoryEntry` proto message.
 
@@ -1781,6 +1897,38 @@ async def delete_fragment(self, fragment_id: str) -> None:
         request = DeleteFragmentRequest(id=fragment_id)
         await self._app_client.DeleteFragment(request, metadata=self._metadata)
 
+    async def get_fragment_history(
+        self, id: str, page_token: Optional[str] = "", page_limit: Optional[int] = 10
+    ) -> List[FragmentHistoryEntry]:
+        """Get fragment history.
+
+        ::
+
+            fragment_history = await cloud.get_fragment_history(
+                id = "12a12ab1-1234-5678-abcd-abcd01234567",
+                page_token = "pg-token",
+                page_limit = 10
+            )
+
+        Args:
+            id (str): ID of the fragment to fetch history for.
+            page_token (Optional[str]): the page token for the fragment history collection
+            page_limit (Optional[int]): the number of fragment history documents to return in the result.
+                The default page limit is 10.
+
+        Raises:
+            GRPCError: if an invalid fragment id, page token or page limit is passed.
+
+        Returns:
+            viam.app.app_client.FragmentHistoryResponse: The fragment history document(s).
+
+        For more information, see `Fleet Management API <https://docs.viam.com/appendix/apis/fleet/>`_.
+        """
+
+        request: GetFragmentHistoryRequest = GetFragmentHistoryRequest(id=id, page_token=page_token, page_limit=page_limit)
+        response: GetFragmentHistoryResponse = await self._app_client.GetFragmentHistory(request, metadata=self._metadata)
+        return [FragmentHistoryEntry.from_proto(fragment_history) for fragment_history in response.history]
+
     async def add_role(
         self,
         org_id: str,

tests/mocks/services.py

@@ -5,6 +5,7 @@
 from numpy.typing import NDArray
 
 from viam.app.data_client import DataClient
+from viam.gen.app.v1.app_pb2 import FragmentHistoryEntry, GetFragmentHistoryRequest, GetFragmentHistoryResponse
 from viam.media.video import ViamImage
 from viam.proto.app import (
     AddRoleRequest,
@@ -1173,6 +1174,7 @@ def __init__(
         available: bool,
         location_auth: LocationAuth,
         robot_part_history: List[RobotPartHistoryEntry],
+        fragment_history: List[FragmentHistoryEntry],
         authorizations: List[Authorization],
         url: str,
         module: Module,
@@ -1195,6 +1197,7 @@ def __init__(
         self.available = available
         self.location_auth = location_auth
         self.robot_part_history = robot_part_history
+        self.fragment_history = fragment_history
         self.authorizations = authorizations
         self.url = url
         self.module = module
@@ -1490,6 +1493,12 @@ async def GetFragment(self, stream: Stream[GetFragmentRequest, GetFragmentRespon
         self.fragment_id = request.id
         await stream.send_message(GetFragmentResponse(fragment=self.fragment))
 
+    async def GetFragmentHistory(self, stream: Stream[GetFragmentHistoryRequest, GetFragmentHistoryResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        self.id = request.id
+        await stream.send_message(GetFragmentHistoryResponse(history=self.fragment_history))
+
     async def CreateFragment(self, stream: Stream[CreateFragmentRequest, CreateFragmentResponse]) -> None:
         request = await stream.recv_message()
         assert request is not None

tests/package.json

@@ -0,0 +1,6 @@
+{
+  "name": "tests",
+  "version": "1.0.0",
+  "dependencies": {
+  }
+}

tests/test_app_client.py

@@ -4,9 +4,10 @@
 from grpclib.testing import ChannelFor
 
 from viam.app.app_client import APIKeyAuthorization, AppClient, Fragment, FragmentVisibilityPB
-from viam.proto.app import APIKey, APIKeyWithAuthorizations, Authorization, AuthorizationDetails, AuthorizedPermissions
+from viam.proto.app import APIKey, APIKeyWithAuthorizations, AuthenticatorInfo, Authorization, AuthorizationDetails, AuthorizedPermissions
 from viam.proto.app import Fragment as FragmentPB
 from viam.proto.app import (
+    FragmentHistoryEntry,
     Location,
     LocationAuth,
     Model,
@@ -36,6 +37,8 @@
 IDS = [ID]
 NAME = "name"
 CID = "cid"
+PAGE_TOKEN = ""
+PAGE_LIMIT = 10
 TIME = datetime_to_timestamp(datetime.now())
 PUBLIC_NAMESPACE = "public_namespace"
 DEFAULT_REGION = "default_region"
@@ -122,6 +125,9 @@
 PART = "part"
 ROBOT_PART_HISTORY_ENTRY = RobotPartHistoryEntry(part=PART, robot=ID, when=TIME, old=None)
 ROBOT_PART_HISTORY = [ROBOT_PART_HISTORY_ENTRY]
+AUTHENTICATOR_INFO = AuthenticatorInfo(value="value", is_deactivated=True, type=1)
+FRAGMENT_HISTORY_ENTRY = FragmentHistoryEntry(fragment=ID, edited_by=AUTHENTICATOR_INFO, old=FRAGMENT, edited_on=TIME)
+FRAGMENT_HISTORY = [FRAGMENT_HISTORY_ENTRY]
 TYPE = "robot"
 ROLE = "operator"
 API_KEY = "key"
@@ -210,6 +216,7 @@ def service() -> MockApp:
         available=AVAILABLE,
         location_auth=LOCATION_AUTH,
         robot_part_history=ROBOT_PART_HISTORY,
+        fragment_history=FRAGMENT_HISTORY,
         authorizations=AUTHORIZATIONS,
         url=URL,
         module=MODULE,
@@ -630,6 +637,16 @@ async def test_delete_fragment(self, service: MockApp):
             await client.delete_fragment(fragment_id=ID)
             assert service.id == ID
 
+    @pytest.mark.asyncio
+    async def test_get_fragment_history(self, service: MockApp):
+        async with ChannelFor([service]) as channel:
+            client = AppClient(channel, METADATA, ID)
+            fragment_history = await client.get_fragment_history(id=ID, page_token="", page_limit=20)
+            assert service.fragment.id == ID
+            assert len(fragment_history) == len(FRAGMENT_HISTORY)
+            for i in range(len(FRAGMENT_HISTORY)):
+                assert fragment_history[i].proto == FRAGMENT_HISTORY[i]
+
     @pytest.mark.asyncio
     async def test_add_role(self, service: MockApp):
         async with ChannelFor([service]) as channel: