[RSDK-8126]: Update fragment methods to use fragment visibility (#677)

src/viam/app/app_client.py

@@ -1,5 +1,6 @@
 import json
 from datetime import datetime
+from enum import Enum
 from typing import Any, AsyncIterator, List, Literal, Mapping, Optional, Tuple, Union
 
 from grpclib.client import Channel
@@ -48,6 +49,7 @@
     DeleteRobotRequest,
 )
 from viam.proto.app import Fragment as FragmentPB
+from viam.proto.app import FragmentVisibility as FragmentVisibilityPB
 from viam.proto.app import (
     GetFragmentRequest,
     GetFragmentResponse,
@@ -284,6 +286,50 @@ class Fragment:
     Use this class to make the attributes of a `viam.proto.app.RobotPart` more accessible and easier to read/interpret.
     """
 
+    class Visibility(str, Enum):
+        """
+        FragmentVisibility specifies who is permitted to view the fragment.
+        """
+
+        PRIVATE = "private"
+        """
+        Only visible to members in the fragment's organization.
+        """
+
+        PUBLIC = "public"
+        """
+        Visible to anyone and appears on the fragments page.
+        """
+
+        PUBLIC_UNLISTED = "public_unlisted"
+        """
+        Visible to anyone but does not appear on the fragments page.
+        """
+
+        UNSPECIFIED = "unspecified"
+        """
+        Uninitialized visibility.
+        """
+
+        @classmethod
+        def from_proto(cls, visibility: FragmentVisibilityPB.ValueType) -> "Fragment.Visibility":
+            if visibility == FragmentVisibilityPB.FRAGMENT_VISIBILITY_PRIVATE:
+                return Fragment.Visibility.PRIVATE
+            if visibility == FragmentVisibilityPB.FRAGMENT_VISIBILITY_PUBLIC:
+                return Fragment.Visibility.PUBLIC
+            if visibility == FragmentVisibilityPB.FRAGMENT_VISIBILITY_PUBLIC_UNLISTED:
+                return Fragment.Visibility.PUBLIC_UNLISTED
+            return Fragment.Visibility.UNSPECIFIED
+
+        def to_proto(self) -> FragmentVisibilityPB.ValueType:
+            if self == self.PRIVATE:
+                return FragmentVisibilityPB.FRAGMENT_VISIBILITY_PRIVATE
+            if self == self.PUBLIC:
+                return FragmentVisibilityPB.FRAGMENT_VISIBILITY_PUBLIC
+            if self == self.PUBLIC_UNLISTED:
+                return FragmentVisibilityPB.FRAGMENT_VISIBILITY_PUBLIC_UNLISTED
+            return FragmentVisibilityPB.FRAGMENT_VISIBILITY_UNSPECIFIED
+
     @classmethod
     def from_proto(cls, fragment: FragmentPB) -> Self:
         """Create a `Fragment` from the .proto defined `Fragment`.
@@ -305,6 +351,7 @@ def from_proto(cls, fragment: FragmentPB) -> Self:
         self.robot_part_count = fragment.robot_part_count
         self.organization_count = fragment.organization_count
         self.only_used_by_owner = fragment.only_used_by_owner
+        self.visibility = Fragment.Visibility.from_proto(fragment.visibility)
         return self
 
     id: str
@@ -317,6 +364,7 @@ def from_proto(cls, fragment: FragmentPB) -> Self:
     robot_part_count: int
     organization_count: int
     only_used_by_owner: bool
+    visibility: Visibility
 
     @property
     def proto(self) -> FragmentPB:
@@ -331,6 +379,7 @@ def proto(self) -> FragmentPB:
             robot_part_count=self.robot_part_count,
             organization_count=self.organization_count,
             only_used_by_owner=self.only_used_by_owner,
+            visibility=self.visibility.to_proto(),
         )
 
 
@@ -1581,25 +1630,36 @@ async def delete_robot(self, robot_id: str) -> None:
         request = DeleteRobotRequest(id=robot_id)
         await self._app_client.DeleteRobot(request, metadata=self._metadata)
 
-    async def list_fragments(self, org_id: str, show_public: bool = True) -> List[Fragment]:
+    async def list_fragments(
+        self, org_id: str, show_public: bool = True, visibilities: Optional[List[Fragment.Visibility]] = None
+    ) -> List[Fragment]:
         """Get a list of fragments under the currently authed-to organization.
 
         ::
 
-            fragments_list = await cloud.list_fragments(org_id="org-id", show_public=False)
+            fragments_list = await cloud.list_fragments(org_id="org-id", visibilities=[])
 
         Args:
             org_id (str): The ID of the organization to list fragments for.
                 You can obtain your organization ID from the Viam app's organization settings page.
-            show_public: Optional boolean specifying whether or not to only show public fragments. If True, only public fragments will
-                return. If False, only private fragments will return. Defaults to True.
+            show_public (bool): **Deprecated**: Use ``visibilities`` instead. Optional boolean specifying whether or not to only show public
+                fragments. If True, only public fragments will return. If False, only private fragments will return. Defaults to True.
+
+                .. deprecated:: 0.25.0
+                    Use ``visibilities`` instead.
+            visibilities (Optional[List[Fragment.Visibility]]): List of FragmentVisibilities specifying which types of fragments to include
+                in the results. If empty, by default only public fragments will be returned.
 
         Returns:
             List[viam.app.app_client.Fragment]: The list of fragments.
 
         For more information, see `Fleet Management API <https://docs.viam.com/appendix/apis/fleet/>`_.
         """
-        request = ListFragmentsRequest(organization_id=org_id, show_public=show_public)
+        request = ListFragmentsRequest(
+            organization_id=org_id,
+            fragment_visibility=map(Fragment.Visibility.to_proto, visibilities if visibilities else []),
+            show_public=show_public,
+        )
         response: ListFragmentsResponse = await self._app_client.ListFragments(request, metadata=self._metadata)
         return [Fragment.from_proto(fragment=fragment) for fragment in response.fragments]
 
@@ -1655,7 +1715,12 @@ async def create_fragment(self, org_id: str, name: str, config: Optional[Mapping
         return Fragment.from_proto(response.fragment)
 
     async def update_fragment(
-        self, fragment_id: str, name: str, config: Optional[Mapping[str, Any]] = None, public: Optional[bool] = None
+        self,
+        fragment_id: str,
+        name: str,
+        config: Optional[Mapping[str, Any]] = None,
+        public: Optional[bool] = None,
+        visibility: Optional[Fragment.Visibility] = None,
     ) -> Fragment:
         """Update a fragment name AND its config and/or visibility.
 
@@ -1670,8 +1735,14 @@ async def update_fragment(
             name (str): New name to associate with the fragment.
             config (Optional[Mapping[str, Any]]): Optional Dictionary representation of new config to assign to specified fragment. Not
                 passing this parameter will leave the fragment's config unchanged.
-            public (bool): Boolean specifying whether the fragment is public. Not passing this parameter will leave the fragment's
-                visibility unchanged. A fragment is private by default when created.
+            public (bool): **Deprecated**: Use ``visibility`` instead. Boolean specifying whether the fragment is public. Not passing this
+                parameter will leave the fragment's visibility unchanged. A fragment is private by default when created.
+
+                .. deprecated:: 0.25.0
+                    Use ``visibility`` instead.
+            visibility (Optional[FragmentVisibility]): Optional FragmentVisibility list specifying who should be allowed
+                to view the fragment. Not passing this parameter will leave the fragment's visibility unchanged.
+                A fragment is private by default when created.
 
         Raises:
             GRPCError: if an invalid ID, name, or config is passed.
@@ -1681,7 +1752,13 @@ async def update_fragment(
 
         For more information, see `Fleet Management API <https://docs.viam.com/appendix/apis/fleet/>`_.
         """
-        request = UpdateFragmentRequest(id=fragment_id, name=name, config=dict_to_struct(config) if config else None, public=public)
+        request = UpdateFragmentRequest(
+            id=fragment_id,
+            name=name,
+            config=dict_to_struct(config) if config else None,
+            public=public,
+            visibility=visibility.to_proto() if visibility else None,
+        )
         response: UpdateFragmentResponse = await self._app_client.UpdateFragment(request, metadata=self._metadata)
         return Fragment.from_proto(response.fragment)
 

tests/mocks/services.py

@@ -1481,7 +1481,7 @@ async def DeleteRobot(self, stream: Stream[DeleteRobotRequest, DeleteRobotRespon
     async def ListFragments(self, stream: Stream[ListFragmentsRequest, ListFragmentsResponse]) -> None:
         request = await stream.recv_message()
         assert request is not None
-        self.show_public = request.show_public
+        self.fragment_visibility = request.fragment_visibility
         await stream.send_message(ListFragmentsResponse(fragments=[self.fragment]))
 
     async def GetFragment(self, stream: Stream[GetFragmentRequest, GetFragmentResponse]) -> None:

tests/test_app_client.py

@@ -3,14 +3,14 @@
 import pytest
 from grpclib.testing import ChannelFor
 
-from viam.app.app_client import APIKeyAuthorization, AppClient
+from viam.app.app_client import APIKeyAuthorization, AppClient, Fragment, FragmentVisibilityPB
 from viam.proto.app import (
     APIKey,
     APIKeyWithAuthorizations,
     Authorization,
     AuthorizationDetails,
     AuthorizedPermissions,
-    Fragment,
+    Fragment as FragmentPB,
     Location,
     LocationAuth,
     Model,
@@ -103,12 +103,13 @@
 LOG_ENTRY = LogEntry(host=HOST, level=LEVEL, time=TIME, logger_name=LOGGER_NAME, message=MESSAGE, caller=None, stack=STACK, fields=None)
 LOG_ENTRIES = [LOG_ENTRY]
 ROBOT_CONFIG = {"key": "value"}
-SHOW_PUBLIC = True
+FRAGMENT_VISIBILITY = [Fragment.Visibility.PUBLIC]
+FRAGMENT_VISIBILITY_PB = [FragmentVisibilityPB.FRAGMENT_VISIBILITY_PUBLIC]
 ORGANIZATION_OWNER = "organization_owner"
 PUBLIC = True
 ORGANIZATION_NAME = "organization_name"
 ONLY_USED_BY_OWNER = True
-FRAGMENT = Fragment(
+FRAGMENT = FragmentPB(
     id=ID,
     name=NAME,
     fragment=None,
@@ -596,8 +597,8 @@ async def test_delete_robot(self, service: MockApp):
     async def test_list_fragments(self, service: MockApp):
         async with ChannelFor([service]) as channel:
             client = AppClient(channel, METADATA, ID)
-            fragments = await client.list_fragments(org_id=ID, show_public=SHOW_PUBLIC)
-            assert service.show_public == SHOW_PUBLIC
+            fragments = await client.list_fragments(org_id=ID, visibilities=FRAGMENT_VISIBILITY)
+            assert service.fragment_visibility == FRAGMENT_VISIBILITY_PB
             assert [fragment.proto for fragment in fragments] == [FRAGMENT]
 
     @pytest.mark.asyncio