[RSDK-10139] Implement User-defined Metadata Methods

src/viam/app/app_client.py

@@ -54,12 +54,16 @@
     GetFragmentResponse,
     GetLocationRequest,
     GetLocationResponse,
+    GetLocationMetadataRequest,
+    GetLocationMetadataResponse,
     GetModuleRequest,
     GetModuleResponse,
     GetOrganizationNamespaceAvailabilityRequest,
     GetOrganizationNamespaceAvailabilityResponse,
     GetOrganizationRequest,
     GetOrganizationResponse,
+    GetOrganizationMetadataRequest,
+    GetOrganizationMetadataResponse,
     GetOrganizationsWithAccessToLocationRequest,
     GetOrganizationsWithAccessToLocationResponse,
     GetRegistryItemRequest,
@@ -70,12 +74,16 @@
     GetRobotPartHistoryResponse,
     GetRobotPartLogsRequest,
     GetRobotPartLogsResponse,
+    GetRobotPartMetadataRequest,
+    GetRobotPartMetadataResponse,
     GetRobotPartRequest,
     GetRobotPartResponse,
     GetRobotPartsRequest,
     GetRobotPartsResponse,
     GetRobotRequest,
     GetRobotResponse,
+    GetRobotMetadataRequest,
+    GetRobotMetadataResponse,
     GetRoverRentalRobotsRequest,
     GetRoverRentalRobotsResponse,
     GetUserIDByEmailRequest,
@@ -134,15 +142,23 @@
     UnshareLocationRequest,
     UpdateFragmentRequest,
     UpdateFragmentResponse,
+    UpdateLocationMetadataRequest,
+    UpdateLocationMetadataResponse,
     UpdateLocationRequest,
     UpdateLocationResponse,
     UpdateModuleRequest,
     UpdateModuleResponse,
     UpdateOrganizationInviteAuthorizationsRequest,
     UpdateOrganizationInviteAuthorizationsResponse,
+    UpdateOrganizationMetadataRequest,
+    UpdateOrganizationMetadataResponse,
     UpdateOrganizationRequest,
     UpdateOrganizationResponse,
     UpdateRegistryItemRequest,
+    UpdateRobotMetadataRequest,
+    UpdateRobotMetadataResponse,
+    UpdateRobotPartMetadataRequest,
+    UpdateRobotPartMetadataResponse,
     UpdateRobotPartRequest,
     UpdateRobotPartResponse,
     UpdateRobotRequest,
@@ -2523,3 +2539,135 @@ async def rotate_key(self, id: str) -> Tuple[str, str]:
         request = RotateKeyRequest(id=id)
         response: RotateKeyResponse = await self._app_client.RotateKey(request, metadata=self._metadata)
         return response.key, response.id
+
+    async def get_organization_metadata(self, org_id: str) -> Mapping[str, Any]:
+        """Get an organization's user-defined metadata.
+
+        ::
+
+            metadata = await cloud.get_organization_metadata(org_id="<YOUR-ORG-ID>")
+
+        Args:
+            org_id (str): The ID of the organization with which the user-defined metadata is associated.
+                You can obtain your organization ID from the Viam app's organization settings page.
+
+        Returns:
+            Mapping[str, Any]: The user-defined metadata converted from JSON to a Python dictionary
+        """
+        request = GetOrganizationMetadataRequest(organization_id=org_id)
+        response: GetOrganizationMetadataResponse = await self._app_client.GetOrganizationMetadata(request)
+        return struct_to_dict(response.data)
+
+    async def update_organization_metadata(self, org_id: str, metadata: Mapping[str, Any]) -> None:
+        """Update an organization's user-defined metadata.
+
+        ::
+
+             await cloud.update_organization_metadata(org_id="<YOUR-ORG-ID>", metadata=)
+
+        Args:
+            organization_id (str): The ID of the organization with which to associate the user-defined metadata.
+                You can obtain your organization ID from the Viam app's organization settings page.
+            metadata (Mapping[str, Any]): The user-defined metadata to upload as a Python dictionary.
+        """
+        request = UpdateOrganizationMetadataRequest(organization_id=org_id, data=dict_to_struct(metadata))
+        _: UpdateOrganizationMetadataResponse = await self._app_client.UpdateOrganizationMetadata(request)
+
+    async def get_location_metadata(self, location_id: str) -> Mapping[str, Any]:
+        """Get a location's user-defined metadata.
+
+        ::
+
+            metadata = await cloud.get_location_metadata(location_id="<YOUR-LOCATION-ID>")
+
+        Args:
+            location_id (str): The ID of the location with which the user-defined metadata is associated.
+                You can obtain your location ID from the Viam app's locations page.
+
+        Returns:
+            Mapping[str, Any]: The user-defined metadata converted from JSON to a Python dictionary.
+        """
+        request = GetLocationMetadataRequest(location_id=location_id)
+        response: GetLocationMetadataResponse = await self._app_client.GetLocationMetadata(request)
+        return struct_to_dict(response.data)
+
+    async def update_location_metadata(self, location_id: str, metadata: Mapping[str, Any]) -> None:
+        """Update a location's user-defined metadata.
+
+        ::
+
+             await cloud.update_location_metadata(location_id="<YOUR-LOCATION-ID>", metadata=)
+
+        Args:
+            location_id (str): The ID of the location with which to associate the user-defined metadata.
+                You can obtain your location ID from the Viam app's locations page.
+            metadata (Mapping[str, Any]): The user-defined metadata converted from JSON to a Python dictionary.
+        """
+        request = UpdateLocationMetadataRequest(location_id=location_id, data=dict_to_struct(metadata))
+        _: UpdateLocationMetadataResponse = await self._app_client.UpdateLocationMetadata(request)
+
+    async def get_robot_metadata(self, robot_id: str) -> Mapping[str, Any]:
+        """Get a robot's user-defined metadata.
+
+        ::
+
+            metadata = await cloud.get_robot_metadata(robot_id="<YOUR-ROBOT-ID>")
+
+        Args:
+            robot_id (str): The ID of the robot with which the user-defined metadata is associated.
+                You can obtain your robot ID from the Viam app's machine page.
+
+        Returns:
+            Mapping[str, Any]: The user-defined metadata converted from JSON to a Python dictionary.
+        """
+        request = GetRobotMetadataRequest(id=robot_id)
+        response: GetRobotMetadataResponse = await self._app_client.GetRobotMetadata(request)
+        return struct_to_dict(response.data)
+
+    async def update_robot_metadata(self, robot_id: str, metadata: Mapping[str, Any]) -> None:
+        """Update a robot's user-defined metadata.
+
+        ::
+
+             await cloud.update_robot_metadata(robot_id="<YOUR-ROBOT-ID>", metadata=)
+
+        Args:
+            robot_id (str): The ID of the robot with which to associate the user-defined metadata.
+                You can obtain your robot ID from the Viam app's machine page.
+            metadata (Mapping[str, Any]): The user-defined metadata converted from JSON to a Python dictionary.
+        """
+        request = UpdateRobotMetadataRequest(id=robot_id, data=dict_to_struct(metadata))
+        _: UpdateRobotMetadataResponse = await self._app_client.UpdateRobotMetadata(request)
+
+    async def get_robot_part_metadata(self, robot_part_id: str) -> Mapping[str, Any]:
+        """Get a robot part's user-defined metadata.
+
+        ::
+
+            metadata = await cloud.get_robot_part_metadata(robot_part_id="<YOUR-ROBOT-PART-ID>")
+
+        Args:
+            robot_part_id (str): The ID of the robot part with which the user-defined metadata is associated.
+                You can obtain your robot part ID from the Viam app's machine page.
+
+        Returns:
+            Mapping[str, Any]: The user-defined metadata converted from JSON to a Python dictionary.
+        """
+        request = GetRobotPartMetadataRequest(id=robot_part_id)
+        response: GetRobotPartMetadataResponse = await self._app_client.GetRobotPartMetadata(request)
+        return struct_to_dict(response.data)
+
+    async def update_robot_part_metadata(self, robot_part_id: str, metadata: Mapping[str, Any]) -> None:
+        """Update a robot part's user-defined metadata.
+
+        ::
+
+             await cloud.update_robot_part_metadata(robot_part_id="<YOUR-ROBOT-PART-ID>", metadata=)
+
+        Args:
+            robot_id (str): The ID of the robot part with which to associate the user-defined metadata.
+                You can obtain your robot part ID from the Viam app's machine page.
+            metadata (Mapping[str, Any]): The user-defined metadata converted from JSON to a Python dictionary.
+        """
+        request = UpdateRobotPartMetadataRequest(id=robot_part_id, data=dict_to_struct(metadata))
+        _: UpdateRobotPartMetadataResponse = await self._app_client.UpdateRobotPartMetadata(request)

tests/mocks/services.py

@@ -63,6 +63,8 @@
     Fragment,
     GetFragmentRequest,
     GetFragmentResponse,
+    GetLocationMetadataRequest,
+    GetLocationMetadataResponse,
     GetLocationRequest,
     GetLocationResponse,
     GetModuleRequest,
@@ -71,16 +73,22 @@
     GetOrganizationNamespaceAvailabilityResponse,
     GetOrganizationRequest,
     GetOrganizationResponse,
+    GetOrganizationMetadataRequest,
+    GetOrganizationMetadataResponse,
     GetOrganizationsWithAccessToLocationRequest,
     GetOrganizationsWithAccessToLocationResponse,
     GetRegistryItemRequest,
     GetRegistryItemResponse,
     GetRobotAPIKeysRequest,
     GetRobotAPIKeysResponse,
+    GetRobotMetadataRequest,
+    GetRobotMetadataResponse,
     GetRobotPartHistoryRequest,
     GetRobotPartHistoryResponse,
     GetRobotPartLogsRequest,
     GetRobotPartLogsResponse,
+    GetRobotPartMetadataRequest,
+    GetRobotPartMetadataResponse,
     GetRobotPartRequest,
     GetRobotPartResponse,
     GetRobotPartsRequest,
@@ -149,16 +157,24 @@
     UnshareLocationResponse,
     UpdateFragmentRequest,
     UpdateFragmentResponse,
+    UpdateLocationMetadataRequest,
+    UpdateLocationMetadataResponse,
     UpdateLocationRequest,
     UpdateLocationResponse,
     UpdateModuleRequest,
     UpdateModuleResponse,
     UpdateOrganizationInviteAuthorizationsRequest,
     UpdateOrganizationInviteAuthorizationsResponse,
+    UpdateOrganizationMetadataRequest,
+    UpdateOrganizationMetadataResponse,
     UpdateOrganizationRequest,
     UpdateOrganizationResponse,
     UpdateRegistryItemRequest,
     UpdateRegistryItemResponse,
+    UpdateRobotMetadataRequest,
+    UpdateRobotMetadataResponse,
+    UpdateRobotPartMetadataRequest,
+    UpdateRobotPartMetadataResponse,
     UpdateRobotPartRequest,
     UpdateRobotPartResponse,
     UpdateRobotRequest,
@@ -1240,6 +1256,10 @@ def __init__(
         self.items = items
         self.package_type = package_type
         self.send_email_invite = False
+        self.organization_metadata = {}
+        self.location_metadata = {}
+        self.robot_metadata = {}
+        self.robot_part_metadata = {}
 
     async def GetUserIDByEmail(self, stream: Stream[GetUserIDByEmailRequest, GetUserIDByEmailResponse]) -> None:
         request = await stream.recv_message()
@@ -1700,6 +1720,49 @@ async def GetRegistryItem(self, stream: Stream[GetRegistryItemRequest, GetRegist
         assert request is not None
         await stream.send_message(GetRegistryItemResponse(item=self.items[0]))
 
+    async def GetOrganizationMetadata(self, stream: Stream[GetOrganizationMetadataRequest, GetOrganizationMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        await stream.send_message(GetOrganizationMetadataResponse(data=self.organization_metadata.get(request.organization_id, dict_to_struct({}))))
+
+    async def UpdateOrganizationMetadata(self, stream: Stream[UpdateOrganizationMetadataRequest, UpdateOrganizationMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        self.organization_metadata[request.organization_id] = request.data
+        await stream.send_message(UpdateOrganizationMetadataResponse())
+
+    async def GetLocationMetadata(self, stream: Stream[GetLocationMetadataRequest, GetLocationMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        await stream.send_message(GetLocationMetadataResponse(data=self.location_metadata.get(request.location_id, dict_to_struct({}))))
+
+    async def UpdateLocationMetadata(self, stream: Stream[UpdateLocationMetadataRequest, UpdateLocationMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        self.location_metadata[request.location_id] = request.data
+        await stream.send_message(UpdateLocationMetadataResponse())
+
+    async def GetRobotMetadata(self, stream: Stream[GetRobotMetadataRequest, GetRobotMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        await stream.send_message(GetRobotMetadataResponse(data=self.robot_metadata.get(request.id, dict_to_struct({}))))
+
+    async def UpdateRobotMetadata(self, stream: Stream[UpdateRobotMetadataRequest, UpdateRobotMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        self.robot_metadata[request.id] = request.data
+        await stream.send_message(UpdateRobotMetadataResponse())
+
+    async def GetRobotPartMetadata(self, stream: Stream[GetRobotPartMetadataRequest, GetRobotPartMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        await stream.send_message(GetRobotPartMetadataResponse(data=self.robot_part_metadata.get(request.id, dict_to_struct({}))))
+
+    async def UpdateRobotPartMetadata(self, stream: Stream[UpdateRobotPartMetadataRequest, UpdateRobotPartMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        self.robot_part_metadata[request.id] = request.data
+        await stream.send_message(UpdateRobotPartMetadataResponse())
 
 class MockGenericService(GenericService):
     timeout: Optional[float] = None

tests/test_app_client.py

@@ -205,6 +205,7 @@
 PLATFORM = "platform"
 MODULE_FILE_INFO = ModuleFileInfo(module_id=ID, version=VERSION, platform=PLATFORM)
 FILE = b"file"
+USER_DEFINED_METADATA = {"number": 0, "string": "string"}
 
 
 @pytest.fixture(scope="function")
@@ -777,3 +778,44 @@ async def test_rotate_key(self, service: MockApp):
             key, id = await client.rotate_key(ID)
             assert key == API_KEY
             assert id == ID
+
+    async def test_get_and_update_organization_metadata(self, service: MockApp):
+        async with ChannelFor([service]) as channel:
+            client = AppClient(channel, METADATA, ID)
+            user_defined_metadata = await client.get_organization_metadata(ID)
+            assert len(user_defined_metadata) == 0
+
+            await client.update_organization_metadata(ID, USER_DEFINED_METADATA)
+            user_defined_metadata = await client.get_organization_metadata(ID)
+            assert user_defined_metadata == USER_DEFINED_METADATA
+
+    async def test_get_and_update_location_metadata(self, service: MockApp):
+        async with ChannelFor([service]) as channel:
+            client = AppClient(channel, METADATA, ID)
+            user_defined_metadata = await client.get_location_metadata(ID)
+            assert len(user_defined_metadata) == 0
+
+
+            await client.update_location_metadata(ID, USER_DEFINED_METADATA)
+            user_defined_metadata = await client.get_location_metadata(ID)
+            assert user_defined_metadata == USER_DEFINED_METADATA
+
+    async def test_get_and_update_robot_metadata(self, service: MockApp):
+        async with ChannelFor([service]) as channel:
+            client = AppClient(channel, METADATA, ID)
+            user_defined_metadata = await client.get_robot_metadata(ID)
+            assert len(user_defined_metadata) == 0
+
+            await client.update_robot_metadata(ID, USER_DEFINED_METADATA)
+            user_defined_metadata = await client.get_robot_metadata(ID)
+            assert user_defined_metadata == USER_DEFINED_METADATA
+
+    async def test_get_and_update_robot_part_metadata(self, service: MockApp):
+        async with ChannelFor([service]) as channel:
+            client = AppClient(channel, METADATA, ID)
+            user_defined_metadata = await client.get_robot_part_metadata(ID)
+            assert len(user_defined_metadata) == 0
+
+            await client.update_robot_part_metadata(ID, USER_DEFINED_METADATA)
+            user_defined_metadata = await client.get_robot_part_metadata(ID)
+            assert user_defined_metadata == USER_DEFINED_METADATA