style: mypy and black

tableauserverclient/server/endpoint/resource_tagger.py

@@ -1,6 +1,6 @@
 import abc
 import copy
-from typing import Generic, Iterable, Set, TypeVar, Union
+from typing import Generic, Iterable, Optional, Protocol, Set, TypeVar, Union, runtime_checkable
 import urllib.parse
 
 from tableauserverclient.server.endpoint.endpoint import Endpoint
@@ -53,15 +53,15 @@ def update_tags(self, baseurl, resource_item):
         logger.info("Updated tags to {0}".format(resource_item.tags))
 
 
-T = TypeVar("T")
+@runtime_checkable
+class Taggable(Protocol):
+    _initial_tags: Set[str]
+    id: Optional[str] = None
+    tags: Set[str]
 
 
-class TaggingMixin(Generic[T]):
-    @abc.abstractmethod
-    def baseurl(self) -> str:
-        raise NotImplementedError("baseurl must be implemented.")
-
-    def add_tags(self, item: Union[T, str], tags: Union[Iterable[str], str]) -> Set[str]:
+class TaggingMixin:
+    def add_tags(self, item: Union[Taggable, str], tags: Union[Iterable[str], str]) -> Set[str]:
         item_id = getattr(item, "id", item)
 
         if not isinstance(item_id, str):
@@ -72,12 +72,12 @@ def add_tags(self, item: Union[T, str], tags: Union[Iterable[str], str]) -> Set[
         else:
             tag_set = set(tags)
 
-        url = f"{self.baseurl}/{item_id}/tags"
+        url = f"{self.baseurl}/{item_id}/tags"  # type: ignore
         add_req = RequestFactory.Tag.add_req(tag_set)
-        server_response = self.put_request(url, add_req)
-        return TagItem.from_response(server_response.content, self.parent_srv.namespace)
+        server_response = self.put_request(url, add_req)  # type: ignore
+        return TagItem.from_response(server_response.content, self.parent_srv.namespace)  # type: ignore
 
-    def delete_tags(self, item: Union[T, str], tags: Union[Iterable[str], str]) -> None:
+    def delete_tags(self, item: Union[Taggable, str], tags: Union[Iterable[str], str]) -> None:
         item_id = getattr(item, "id", item)
 
         if not isinstance(item_id, str):
@@ -90,10 +90,10 @@ def delete_tags(self, item: Union[T, str], tags: Union[Iterable[str], str]) -> N
 
         for tag in tag_set:
             encoded_tag_name = urllib.parse.quote(tag)
-            url = f"{self.baseurl}/{item_id}/tags/{encoded_tag_name}"
-            self.delete_request(url)
+            url = f"{self.baseurl}/{item_id}/tags/{encoded_tag_name}"  # type: ignore
+            self.delete_request(url)  # type: ignore
 
-    def update_tags(self, item: T) -> None:
+    def update_tags(self, item: Taggable) -> None:
         if item.tags == item._initial_tags:
             return
 

tableauserverclient/server/endpoint/workbooks_endpoint.py

@@ -58,7 +58,7 @@
 PathOrFileW = Union[FilePath, FileObjectW]
 
 
-class Workbooks(QuerysetEndpoint[WorkbookItem], TaggingMixin[WorkbookItem]):
+class Workbooks(QuerysetEndpoint[WorkbookItem], TaggingMixin):
     def __init__(self, parent_srv: "Server") -> None:
         super(Workbooks, self).__init__(parent_srv)
         self._resource_tagger = _ResourceTagger(parent_srv)
@@ -502,6 +502,6 @@ def schedule_extract_refresh(
         return self.parent_srv.schedules.add_to_schedule(schedule_id, workbook=item)
 
 
-Workbooks.add_tags = api(version="1.0")(Workbooks.add_tags)
-Workbooks.delete_tags = api(version="1.0")(Workbooks.delete_tags)
-Workbooks.update_tags = api(version="1.0")(Workbooks.update_tags)
+Workbooks.add_tags = api(version="1.0")(Workbooks.add_tags)  # type: ignore
+Workbooks.delete_tags = api(version="1.0")(Workbooks.delete_tags)  # type: ignore
+Workbooks.update_tags = api(version="1.0")(Workbooks.update_tags)  # type: ignore

test/assets/workbook_add_tags.xml

@@ -0,0 +1,9 @@
+<?xml version='1.0' encoding='UTF-8'?>
+<tsResponse xmlns="http://tableau.com/api" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tableau.com/api http://tableau.com/api/ts-api-2.3.xsd">
+	<tags>
+		<tag label="a" />
+		<tag label="b" />
+		<tag label="c" />
+		<tag label="d" />
+	</tags>
+</tsResponse>

test/test_workbook.py

@@ -18,7 +18,6 @@
 
 TEST_ASSET_DIR = os.path.join(os.path.dirname(__file__), "assets")
 
-ADD_TAG_XML = os.path.join(TEST_ASSET_DIR, "workbook_add_tag.xml")
 ADD_TAGS_XML = os.path.join(TEST_ASSET_DIR, "workbook_add_tags.xml")
 GET_BY_ID_XML = os.path.join(TEST_ASSET_DIR, "workbook_get_by_id.xml")
 GET_BY_ID_XML_PERSONAL = os.path.join(TEST_ASSET_DIR, "workbook_get_by_id_personal.xml")
@@ -895,70 +894,3 @@ def test_odata_connection(self) -> None:
 
         assert xml_connection is not None
         self.assertEqual(xml_connection.get("serverAddress"), url)
-
-    def test_add_tags(self) -> None:
-        workbook = TSC.WorkbookItem("project", "test")
-        workbook._id = "06b944d2-959d-4604-9305-12323c95e70e"
-        tags = list("abcd")
-
-        with requests_mock.mock() as m:
-            m.put(
-                f"{self.baseurl}/{workbook.id}/tags",
-                status_code=200,
-                text=Path(ADD_TAGS_XML).read_text(),
-            )
-            tag_result = self.server.workbooks.add_tags(workbook, tags)
-
-        for a, b in zip(sorted(tag_result), sorted(tags)):
-            self.assertEqual(a, b)
-
-    def test_add_tag(self) -> None:
-        workbook = TSC.WorkbookItem("project", "test")
-        workbook._id = "06b944d2-959d-4604-9305-12323c95e70e"
-        tags = "a"
-
-        with requests_mock.mock() as m:
-            m.put(
-                f"{self.baseurl}/{workbook.id}/tags",
-                status_code=200,
-                text=Path(ADD_TAG_XML).read_text(),
-            )
-            tag_result = self.server.workbooks.add_tags(workbook, tags)
-
-        for a, b in zip(sorted(tag_result), sorted(tags)):
-            self.assertEqual(a, b)
-
-    def test_add_tag_id(self) -> None:
-        workbook = TSC.WorkbookItem("project", "test")
-        workbook._id = "06b944d2-959d-4604-9305-12323c95e70e"
-        tags = "a"
-
-        with requests_mock.mock() as m:
-            m.put(
-                f"{self.baseurl}/{workbook.id}/tags",
-                status_code=200,
-                text=Path(ADD_TAG_XML).read_text(),
-            )
-            tag_result = self.server.workbooks.add_tags(workbook.id, tags)
-
-        for a, b in zip(sorted(tag_result), sorted(tags)):
-            self.assertEqual(a, b)
-
-    def test_delete_tags(self) -> None:
-        workbook = TSC.WorkbookItem("project", "test")
-        workbook._id = "06b944d2-959d-4604-9305-12323c95e70e"
-        tags = list("abcd")
-
-        matcher = re.compile(rf"{self.baseurl}\/{workbook.id}\/tags\/[abcd]")
-        with requests_mock.mock() as m:
-            m.delete(
-                matcher,
-                status_code=200,
-                text="",
-            )
-            self.server.workbooks.delete_tags(workbook, tags)
-            history = m.request_history
-
-        self.assertEqual(len(history), len(tags))
-        urls = sorted([r.url.split("/")[-1] for r in history])
-        self.assertEqual(urls, sorted(tags))