chore: make datasource typing more specific

tableauserverclient/models/datasource_item.py

@@ -490,7 +490,7 @@ def _set_values(
             self._owner = owner
 
     @classmethod
-    def from_response(cls, resp: str, ns: dict) -> list["DatasourceItem"]:
+    def from_response(cls, resp: bytes, ns: dict) -> list["DatasourceItem"]:
         all_datasource_items = list()
         parsed_response = fromstring(resp)
         all_datasource_xml = parsed_response.findall(".//t:datasource", namespaces=ns)

tableauserverclient/server/endpoint/datasources_endpoint.py

@@ -6,8 +6,8 @@
 
 from contextlib import closing
 from pathlib import Path
-from typing import Literal, Optional, TYPE_CHECKING, Union, overload
-from collections.abc import Iterable, Mapping, Sequence
+from typing import Literal, Optional, TYPE_CHECKING, TypedDict, TypeVar, Union, overload
+from collections.abc import Iterable, Sequence
 
 from tableauserverclient.helpers.headers import fix_filename
 from tableauserverclient.models.dqw_item import DQWItem
@@ -50,13 +50,50 @@
 FileObject = Union[io.BufferedReader, io.BytesIO]
 PathOrFile = Union[FilePath, FileObject]
 
-FilePath = Union[str, os.PathLike]
 FileObjectR = Union[io.BufferedReader, io.BytesIO]
 FileObjectW = Union[io.BufferedWriter, io.BytesIO]
 PathOrFileR = Union[FilePath, FileObjectR]
 PathOrFileW = Union[FilePath, FileObjectW]
 
 
+HyperActionCondition = TypedDict(
+    "HyperActionCondition",
+    {
+        "op": str,
+        "target-col": str,
+        "source-col": str,
+    },
+)
+
+HyperActionRow = TypedDict(
+    "HyperActionRow",
+    {
+        "action": Literal[
+            "update",
+            "upsert",
+            "delete",
+        ],
+        "source-table": str,
+        "target-table": str,
+        "condition": HyperActionCondition,
+    },
+)
+
+HyperActionTable = TypedDict(
+    "HyperActionTable",
+    {
+        "action": Literal[
+            "insert",
+            "replace",
+        ],
+        "source-table": str,
+        "target-table": str,
+    },
+)
+
+HyperAction = Union[HyperActionTable, HyperActionRow]
+
+
 class Datasources(QuerysetEndpoint[DatasourceItem], TaggingMixin[DatasourceItem]):
     def __init__(self, parent_srv: "Server") -> None:
         super().__init__(parent_srv)
@@ -191,16 +228,34 @@ def delete(self, datasource_id: str) -> None:
         self.delete_request(url)
         logger.info(f"Deleted single datasource (ID: {datasource_id})")
 
+    T = TypeVar("T", bound=FileObjectW)
+
+    @overload
+    def download(
+        self,
+        datasource_id: str,
+        filepath: T,
+        include_extract: bool = True,
+    ) -> T: ...
+
+    @overload
+    def download(
+        self,
+        datasource_id: str,
+        filepath: Optional[FilePath] = None,
+        include_extract: bool = True,
+    ) -> str: ...
+
     # Download 1 datasource by id
     @api(version="2.0")
     @parameter_added_in(no_extract="2.5")
     @parameter_added_in(include_extract="2.5")
     def download(
         self,
-        datasource_id: str,
-        filepath: Optional[PathOrFileW] = None,
-        include_extract: bool = True,
-    ) -> PathOrFileW:
+        datasource_id,
+        filepath=None,
+        include_extract=True,
+    ):
         """
         Downloads the specified data source from a site. The data source is
         downloaded as a .tdsx file.
@@ -479,13 +534,13 @@ def publish(
     @parameter_added_in(as_job="3.0")
     def publish(
         self,
-        datasource_item: DatasourceItem,
-        file: PathOrFileR,
-        mode: str,
-        connection_credentials: Optional[ConnectionCredentials] = None,
-        connections: Optional[Sequence[ConnectionItem]] = None,
-        as_job: bool = False,
-    ) -> Union[DatasourceItem, JobItem]:
+        datasource_item,
+        file,
+        mode,
+        connection_credentials=None,
+        connections=None,
+        as_job=False,
+    ):
         """
         Publishes a data source to a server, or appends data to an existing
         data source.
@@ -631,7 +686,7 @@ def update_hyper_data(
         datasource_or_connection_item: Union[DatasourceItem, ConnectionItem, str],
         *,
         request_id: str,
-        actions: Sequence[Mapping],
+        actions: Sequence[HyperAction],
         payload: Optional[FilePath] = None,
     ) -> JobItem:
         """
@@ -898,15 +953,35 @@ def _get_datasource_revisions(
         revisions = RevisionItem.from_response(server_response.content, self.parent_srv.namespace, datasource_item)
         return revisions
 
-    # Download 1 datasource revision by revision number
-    @api(version="2.3")
+    T = TypeVar("T", bound=FileObjectW)
+
+    @overload
+    def download_revision(
+        self,
+        datasource_id: str,
+        revision_number: Optional[str],
+        filepath: T,
+        include_extract: bool = True,
+    ) -> T: ...
+
+    @overload
     def download_revision(
         self,
         datasource_id: str,
         revision_number: Optional[str],
-        filepath: Optional[PathOrFileW] = None,
+        filepath: Optional[FilePath] = None,
         include_extract: bool = True,
-    ) -> PathOrFileW:
+    ) -> str: ...
+
+    # Download 1 datasource revision by revision number
+    @api(version="2.3")
+    def download_revision(
+        self,
+        datasource_id,
+        revision_number,
+        filepath=None,
+        include_extract=True,
+    ):
         """
         Downloads a specific version of a data source prior to the current one
         in .tdsx format. To download the current version of a data source set