chore: refactor XML payload into RequestFactory

jorwoods · jacalata · commit 72885d1af48a · 2026-02-03T11:39:48.000-08:00
Also correct the type hints to clarify that it accepts any
Iterable.
diff --git a/tableauserverclient/models/connection_item.py b/tableauserverclient/models/connection_item.py
@@ -84,10 +84,6 @@ def connection_type(self) -> Optional[str]:
     def query_tagging(self) -> Optional[bool]:
         return self._query_tagging
 
-    @property
-    def auth_type(self) -> Optional[str]:
-        return self._auth_type
-
     @query_tagging.setter
     @property_is_boolean
     def query_tagging(self, value: Optional[bool]):
diff --git a/tableauserverclient/server/endpoint/datasources_endpoint.py b/tableauserverclient/server/endpoint/datasources_endpoint.py
@@ -376,8 +376,14 @@ def update_connection(
 
     @api(version="3.26")
     def update_connections(
-        self, datasource_item: DatasourceItem, connection_luids: list[str], authentication_type: str, username: Optional[str] = None, password: Optional[str] = None, embed_password: Optional[bool] = None
-    ) -> list[str]:
+        self,
+        datasource_item: DatasourceItem,
+        connection_luids: Iterable[str],
+        authentication_type: str,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        embed_password: Optional[bool] = None,
+    ) -> Iterable[str]:
         """
         Bulk updates one or more datasource connections by LUID.
 
@@ -386,7 +392,7 @@ def update_connections(
         datasource_item : DatasourceItem
             The datasource item containing the connections.
 
-        connection_luids : list of str
+        connection_luids : Iterable of str
             The connection LUIDs to update.
 
         authentication_type : str
@@ -403,41 +409,23 @@ def update_connections(
 
         Returns
         -------
-        list of str
+        Iterable of str
             The connection LUIDs that were updated.
         """
-        from xml.etree.ElementTree import Element, SubElement, tostring
 
         url = f"{self.baseurl}/{datasource_item.id}/connections"
         print("Method URL:", url)
 
-        ts_request = Element("tsRequest")
-
-        # <connectionLuids>
-        conn_luids_elem = SubElement(ts_request, "connectionLuids")
-        for luid in connection_luids:
-            SubElement(conn_luids_elem, "connectionLuid").text = luid
-
-        # <connection>
-        connection_elem = SubElement(ts_request, "connection")
-        connection_elem.set("authenticationType", authentication_type)
-
-        if username:
-            connection_elem.set("userName", username)
-
-        if password:
-            connection_elem.set("password", password)
-
-        if embed_password is not None:
-            connection_elem.set("embedPassword", str(embed_password).lower())
-
-        request_body = tostring(ts_request)
-
+        request_body = RequestFactory.Datasource.update_connections_req(
+            connection_luids=connection_luids,
+            authentication_type=authentication_type,
+            username=username,
+            password=password,
+            embed_password=embed_password,
+        )
         response = self.put_request(url, request_body)
 
-        logger.info(
-            f"Updated connections for datasource {datasource_item.id}: {', '.join(connection_luids)}"
-        )
+        logger.info(f"Updated connections for datasource {datasource_item.id}: {', '.join(connection_luids)}")
         return connection_luids
 
     @api(version="2.8")
diff --git a/tableauserverclient/server/endpoint/workbooks_endpoint.py b/tableauserverclient/server/endpoint/workbooks_endpoint.py
@@ -330,69 +330,59 @@ def update_connection(self, workbook_item: WorkbookItem, connection_item: Connec
 
     # Update workbook_connections
     @api(version="3.26")
-    def update_connections(self, workbook_item: WorkbookItem, connection_luids: list[str], authentication_type: str, username: Optional[str] = None, password: Optional[str] = None, embed_password: Optional[bool] = None
-     ) -> list[str]:
-         """
-         Bulk updates one or more workbook connections by LUID, including authenticationType, username, password, and embedPassword.
-
-         Parameters
-         ----------
-         workbook_item : WorkbookItem
-             The workbook item containing the connections.
-
-         connection_luids : list of str
-             The connection LUIDs to update.
-
-         authentication_type : str
-             The authentication type to use (e.g., 'AD Service Principal').
-
-         username : str, optional
-             The username to set (e.g., client ID for keypair auth).
-
-         password : str, optional
-             The password or secret to set.
-
-         embed_password : bool, optional
-             Whether to embed the password.
+    def update_connections(
+        self,
+        workbook_item: WorkbookItem,
+        connection_luids: Iterable[str],
+        authentication_type: str,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        embed_password: Optional[bool] = None,
+    ) -> Iterable[str]:
+        """
+        Bulk updates one or more workbook connections by LUID, including authenticationType, username, password, and embedPassword.
 
-         Returns
-         -------
-         list of str
-             The connection LUIDs that were updated.
-         """
-         from xml.etree.ElementTree import Element, SubElement, tostring
+        Parameters
+        ----------
+        workbook_item : WorkbookItem
+            The workbook item containing the connections.
 
-         url = f"{self.baseurl}/{workbook_item.id}/connections"
+        connection_luids : Iterable of str
+            The connection LUIDs to update.
 
-         ts_request = Element("tsRequest")
+        authentication_type : str
+            The authentication type to use (e.g., 'AD Service Principal').
 
-         # <connectionLuids>
-         conn_luids_elem = SubElement(ts_request, "connectionLuids")
-         for luid in connection_luids:
-             SubElement(conn_luids_elem, "connectionLuid").text = luid
+        username : str, optional
+            The username to set (e.g., client ID for keypair auth).
 
-         # <connection>
-         connection_elem = SubElement(ts_request, "connection")
-         connection_elem.set("authenticationType", authentication_type)
+        password : str, optional
+            The password or secret to set.
 
-         if username:
-             connection_elem.set("userName", username)
+        embed_password : bool, optional
+            Whether to embed the password.
 
-         if password:
-             connection_elem.set("password", password)
+        Returns
+        -------
+        Iterable of str
+            The connection LUIDs that were updated.
+        """
 
-         if embed_password is not None:
-             connection_elem.set("embedPassword", str(embed_password).lower())
+        url = f"{self.baseurl}/{workbook_item.id}/connections"
 
-         request_body = tostring(ts_request)
+        request_body = RequestFactory.Workbook.update_connections_req(
+            connection_luids,
+            authentication_type,
+            username=username,
+            password=password,
+            embed_password=embed_password,
+        )
 
-         # Send request
-         response = self.put_request(url, request_body)
+        # Send request
+        response = self.put_request(url, request_body)
 
-         logger.info(
-             f"Updated connections for workbook {workbook_item.id}: {', '.join(connection_luids)}"
-         )
-         return connection_luids
+        logger.info(f"Updated connections for workbook {workbook_item.id}: {', '.join(connection_luids)}")
+        return connection_luids
 
     # Download workbook contents with option of passing in filepath
     @api(version="2.0")
diff --git a/test/test_datasource.py b/test/test_datasource.py
@@ -158,29 +158,27 @@ def test_update_copy_fields(server) -> None:
 
 
     def test_update_connections(self) -> None:
-        populate_xml, response_xml = read_xml_assets(
-            POPULATE_CONNECTIONS_XML,
-            UPDATE_CONNECTIONS_XML
-        )
+        populate_xml, response_xml = read_xml_assets(POPULATE_CONNECTIONS_XML, UPDATE_CONNECTIONS_XML)
 
         with requests_mock.Mocker() as m:
 
             datasource_id = "9dbd2263-16b5-46e1-9c43-a76bb8ab65fb"
-            connection_luids = [
-                "be786ae0-d2bf-4a4b-9b34-e2de8d2d4488",
-                "a1b2c3d4-e5f6-7a8b-9c0d-123456789abc"
-            ]
+            connection_luids = ["be786ae0-d2bf-4a4b-9b34-e2de8d2d4488", "a1b2c3d4-e5f6-7a8b-9c0d-123456789abc"]
 
             datasource = TSC.DatasourceItem(datasource_id)
             datasource._id = "9dbd2263-16b5-46e1-9c43-a76bb8ab65fb"
             datasource.owner_id = "dd2239f6-ddf1-4107-981a-4cf94e415794"
             self.server.version = "3.26"
 
             url = f"{self.server.baseurl}/{datasource.id}/connections"
-            m.get("http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/datasources/9dbd2263-16b5-46e1-9c43-a76bb8ab65fb/connections", text=populate_xml)
-            m.put("http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/datasources/9dbd2263-16b5-46e1-9c43-a76bb8ab65fb/connections", text=response_xml)
-
-
+            m.get(
+                "http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/datasources/9dbd2263-16b5-46e1-9c43-a76bb8ab65fb/connections",
+                text=populate_xml,
+            )
+            m.put(
+                "http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/datasources/9dbd2263-16b5-46e1-9c43-a76bb8ab65fb/connections",
+                text=response_xml,
+            )
 
             print("BASEURL:", self.server.baseurl)
             print("Calling PUT on:", f"{self.server.baseurl}/{datasource.id}/connections")
@@ -191,12 +189,11 @@ def test_update_connections(self) -> None:
                 authentication_type="auth-keypair",
                 username="testuser",
                 password="testpass",
-                embed_password=True
+                embed_password=True,
             )
 
             self.assertEqual(updated_luids, connection_luids)
 
-
     def test_populate_permissions(self) -> None:
         with open(asset(POPULATE_PERMISSIONS_XML), "rb") as f:
             response_xml = f.read().decode("utf-8")
diff --git a/test/test_workbook.py b/test/test_workbook.py
@@ -709,34 +709,32 @@ def test_publish_with_thumbnails_user_id(server: TSC.Server) -> None:
 
 
     def test_update_workbook_connections(self) -> None:
-        populate_xml, response_xml = read_xml_assets(
-                    POPULATE_CONNECTIONS_XML,
-                    UPDATE_CONNECTIONS_XML
-                )
-
+        populate_xml, response_xml = read_xml_assets(POPULATE_CONNECTIONS_XML, UPDATE_CONNECTIONS_XML)
 
         with requests_mock.Mocker() as m:
             workbook_id = "1a2b3c4d-5e6f-7a8b-9c0d-112233445566"
-            connection_luids = [
-                "abc12345-def6-7890-gh12-ijklmnopqrst",
-                "1234abcd-5678-efgh-ijkl-0987654321mn"
-            ]
+            connection_luids = ["abc12345-def6-7890-gh12-ijklmnopqrst", "1234abcd-5678-efgh-ijkl-0987654321mn"]
 
             workbook = TSC.WorkbookItem(workbook_id)
             workbook._id = workbook_id
             self.server.version = "3.26"
             url = f"{self.server.baseurl}/{workbook_id}/connections"
-            m.get("http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks/1a2b3c4d-5e6f-7a8b-9c0d-112233445566/connections", text=populate_xml)
-            m.put("http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks/1a2b3c4d-5e6f-7a8b-9c0d-112233445566/connections", text=response_xml)
-
+            m.get(
+                "http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks/1a2b3c4d-5e6f-7a8b-9c0d-112233445566/connections",
+                text=populate_xml,
+            )
+            m.put(
+                "http://test/api/3.26/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks/1a2b3c4d-5e6f-7a8b-9c0d-112233445566/connections",
+                text=response_xml,
+            )
 
             updated_luids = self.server.workbooks.update_connections(
                 workbook_item=workbook,
                 connection_luids=connection_luids,
                 authentication_type="AD Service Principal",
                 username="svc-client",
                 password="secret-token",
-                embed_password=True
+                embed_password=True,
             )
 
             self.assertEqual(updated_luids, connection_luids)
