chore: refactor XML payload into RequestFactory

Also correct the type hints to clarify that it accepts any
Iterable.

Author: jorwoods

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]):
@@ -99,6 +95,14 @@ def query_tagging(self, value: Optional[bool]):
             return
         self._query_tagging = value
 
+    @property
+    def auth_type(self) -> Optional[str]:
+        return self._auth_type
+
+    @auth_type.setter
+    def auth_type(self, value: Optional[str]):
+        self._auth_type = value
+
     def __repr__(self):
         return "<ConnectionItem#{_id} embed={embed_password} type={_connection_type} auth={_auth_type} username={username}>".format(
             **self.__dict__

tableauserverclient/server/endpoint/datasources_endpoint.py

@@ -321,8 +321,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.
 
@@ -331,7 +337,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
@@ -348,41 +354,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")

tableauserverclient/server/endpoint/workbooks_endpoint.py

@@ -327,69 +327,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")

tableauserverclient/server/request_factory.py

@@ -244,6 +244,32 @@ def publish_req_chunked(self, datasource_item, connection_credentials=None, conn
         parts = {"request_payload": ("", xml_request, "text/xml")}
         return _add_multipart(parts)
 
+    @_tsrequest_wrapped
+    def update_connections_req(
+        self,
+        element: ET.Element,
+        connection_luids: Iterable[str],
+        authentication_type: str,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        embed_password: Optional[bool] = None,
+    ):
+        conn_luids_elem = ET.SubElement(element, "connectionLUIDs")
+        for luid in connection_luids:
+            ET.SubElement(conn_luids_elem, "connectionLUID").text = luid
+
+        connection_elem = ET.SubElement(element, "connection")
+        connection_elem.set("authenticationType", authentication_type)
+
+        if username is not None:
+            connection_elem.set("userName", username)
+
+        if password is not None:
+            connection_elem.set("password", password)
+
+        if embed_password is not None:
+            connection_elem.set("embedPassword", str(embed_password).lower())
+
 
 class DQWRequest:
     def add_req(self, dqw_item):
@@ -1092,6 +1118,32 @@ def embedded_extract_req(
                 if (id_ := datasource_item.id) is not None:
                     datasource_element.attrib["id"] = id_
 
+    @_tsrequest_wrapped
+    def update_connections_req(
+        self,
+        element: ET.Element,
+        connection_luids: Iterable[str],
+        authentication_type: str,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        embed_password: Optional[bool] = None,
+    ):
+        conn_luids_elem = ET.SubElement(element, "connectionLUIDs")
+        for luid in connection_luids:
+            ET.SubElement(conn_luids_elem, "connectionLUID").text = luid
+
+        connection_elem = ET.SubElement(element, "connection")
+        connection_elem.set("authenticationType", authentication_type)
+
+        if username is not None:
+            connection_elem.set("userName", username)
+
+        if password is not None:
+            connection_elem.set("password", password)
+
+        if embed_password is not None:
+            connection_elem.set("embedPassword", str(embed_password).lower())
+
 
 class Connection:
     @_tsrequest_wrapped

test/test_datasource.py

@@ -219,29 +219,27 @@ def test_update_connection(self) -> None:
             self.assertEqual("foo", new_connection.username)
 
     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")
@@ -252,12 +250,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")

test/test_workbook.py

@@ -982,34 +982,32 @@ def test_odata_connection(self) -> None:
         self.assertEqual(xml_connection.get("serverAddress"), url)
 
     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)