chore: replace Union with modern syntax

Author: jorwoods

tableauserverclient/models/favorites_item.py

@@ -25,7 +25,7 @@ class FavoriteType(TypedDict):
 
 class FavoriteItem:
     @classmethod
-    def from_response(cls, xml: Union[str, bytes], namespace: dict) -> FavoriteType:
+    def from_response(cls, xml: str | bytes, namespace: dict) -> FavoriteType:
         favorites: FavoriteType = {
             "collections": [],
             "datasources": [],

tableauserverclient/models/schedule_item.py

@@ -20,7 +20,7 @@
     from requests import Response
 
 
-Interval = Union[HourlyInterval, DailyInterval, WeeklyInterval, MonthlyInterval]
+Interval = HourlyInterval | DailyInterval | WeeklyInterval | MonthlyInterval
 
 
 class ScheduleItem:

tableauserverclient/models/site_item.py

@@ -154,9 +154,9 @@ def __init__(
         notify_site_admins_on_throttle: bool = False,
         authoring_enabled: bool = True,
         custom_subscription_email_enabled: bool = False,
-        custom_subscription_email: Union[str, bool] = False,
+        custom_subscription_email: str | bool = False,
         custom_subscription_footer_enabled: bool = False,
-        custom_subscription_footer: Union[str, bool] = False,
+        custom_subscription_footer: str | bool = False,
         ask_data_mode: str = "EnabledByDefault",
         named_sharing_enabled: bool = True,
         mobile_biometrics_enabled: bool = False,
@@ -555,11 +555,11 @@ def custom_subscription_email_enabled(self, value: bool) -> None:
         self._custom_subscription_email_enabled = value
 
     @property
-    def custom_subscription_email(self) -> Union[str, bool]:
+    def custom_subscription_email(self) -> str | bool:
         return self._custom_subscription_email
 
     @custom_subscription_email.setter
-    def custom_subscription_email(self, value: Union[str, bool]):
+    def custom_subscription_email(self, value: str | bool):
         self._custom_subscription_email = value
 
     @property
@@ -572,11 +572,11 @@ def custom_subscription_footer_enabled(self, value: bool) -> None:
         self._custom_subscription_footer_enabled = value
 
     @property
-    def custom_subscription_footer(self) -> Union[str, bool]:
+    def custom_subscription_footer(self) -> str | bool:
         return self._custom_subscription_footer
 
     @custom_subscription_footer.setter
-    def custom_subscription_footer(self, value: Union[str, bool]) -> None:
+    def custom_subscription_footer(self, value: str | bool) -> None:
         self._custom_subscription_footer = value
 
     @property

tableauserverclient/models/tableau_types.py

@@ -40,7 +40,7 @@ class Resource:
 ]
 
 
-def plural_type(content_type: Union[Resource, str]) -> str:
+def plural_type(content_type: Resource | str) -> str:
     if content_type == Resource.Lens:
         return "lenses"
     else:

tableauserverclient/server/endpoint/custom_views_endpoint.py

@@ -32,12 +32,12 @@
 update the name or owner of a custom view.
 """
 
-FilePath = Union[str, os.PathLike]
-FileObject = Union[io.BufferedReader, io.BytesIO]
-FileObjectR = Union[io.BufferedReader, io.BytesIO]
-FileObjectW = Union[io.BufferedWriter, io.BytesIO]
-PathOrFileR = Union[FilePath, FileObjectR]
-PathOrFileW = Union[FilePath, FileObjectW]
+FilePath = str | os.PathLike
+FileObject = io.BufferedReader | io.BytesIO
+FileObjectR = io.BufferedReader | io.BytesIO
+FileObjectW = io.BufferedWriter | io.BytesIO
+PathOrFileR = FilePath | FileObjectR
+PathOrFileW = FilePath | FileObjectW
 io_types_r = (io.BufferedReader, io.BytesIO)
 io_types_w = (io.BufferedWriter, io.BytesIO)
 

tableauserverclient/server/endpoint/data_alert_endpoint.py

@@ -44,7 +44,7 @@ def get_by_id(self, dataAlert_id: str) -> DataAlertItem:
         return DataAlertItem.from_response(server_response.content, self.parent_srv.namespace)[0]
 
     @api(version="3.2")
-    def delete(self, dataAlert: Union[DataAlertItem, str]) -> None:
+    def delete(self, dataAlert: DataAlertItem | str) -> None:
         if isinstance(dataAlert, DataAlertItem):
             dataAlert_id = dataAlert.id
         elif isinstance(dataAlert, str):
@@ -60,7 +60,7 @@ def delete(self, dataAlert: Union[DataAlertItem, str]) -> None:
         logger.info(f"Deleted single dataAlert (ID: {dataAlert_id})")
 
     @api(version="3.2")
-    def delete_user_from_alert(self, dataAlert: Union[DataAlertItem, str], user: Union[UserItem, str]) -> None:
+    def delete_user_from_alert(self, dataAlert: DataAlertItem | str, user: UserItem | str) -> None:
         if isinstance(dataAlert, DataAlertItem):
             dataAlert_id = dataAlert.id
         elif isinstance(dataAlert, str):
@@ -85,7 +85,7 @@ def delete_user_from_alert(self, dataAlert: Union[DataAlertItem, str], user: Uni
         logger.info(f"Deleted User (ID {user_id}) from dataAlert (ID: {dataAlert_id})")
 
     @api(version="3.2")
-    def add_user_to_alert(self, dataAlert_item: DataAlertItem, user: Union[UserItem, str]) -> UserItem:
+    def add_user_to_alert(self, dataAlert_item: DataAlertItem, user: UserItem | str) -> UserItem:
         if isinstance(user, UserItem):
             user_id = user.id
         elif isinstance(user, str):

tableauserverclient/server/endpoint/databases_endpoint.py

@@ -211,11 +211,11 @@ def delete_dqw(self, item: DatabaseItem) -> None:
         self._data_quality_warnings.clear(item)
 
     @api(version="3.9")
-    def add_tags(self, item: Union[DatabaseItem, str], tags: Iterable[str]) -> set[str]:
+    def add_tags(self, item: DatabaseItem | str, tags: Iterable[str]) -> set[str]:
         return super().add_tags(item, tags)
 
     @api(version="3.9")
-    def delete_tags(self, item: Union[DatabaseItem, str], tags: Iterable[str]) -> None:
+    def delete_tags(self, item: DatabaseItem | str, tags: Iterable[str]) -> None:
         super().delete_tags(item, tags)
 
     @api(version="3.9")

tableauserverclient/server/endpoint/datasources_endpoint.py

@@ -46,14 +46,14 @@
 io_types_r = (io.BytesIO, io.BufferedReader)
 io_types_w = (io.BytesIO, io.BufferedWriter)
 
-FilePath = Union[str, os.PathLike]
-FileObject = Union[io.BufferedReader, io.BytesIO]
-PathOrFile = Union[FilePath, FileObject]
+FilePath = str | os.PathLike
+FileObject = io.BufferedReader | io.BytesIO
+PathOrFile = FilePath | FileObject
 
-FileObjectR = Union[io.BufferedReader, io.BytesIO]
-FileObjectW = Union[io.BufferedWriter, io.BytesIO]
-PathOrFileR = Union[FilePath, FileObjectR]
-PathOrFileW = Union[FilePath, FileObjectW]
+FileObjectR = io.BufferedReader | io.BytesIO
+FileObjectW = io.BufferedWriter | io.BytesIO
+PathOrFileR = FilePath | FileObjectR
+PathOrFileW = FilePath | FileObjectW
 
 
 HyperActionCondition = TypedDict(
@@ -91,7 +91,7 @@
     },
 )
 
-HyperAction = Union[HyperActionTable, HyperActionRow]
+HyperAction = HyperActionTable | HyperActionRow
 
 
 class Datasources(QuerysetEndpoint[DatasourceItem], TaggingMixin[DatasourceItem]):
@@ -434,7 +434,7 @@ def update_connections(
         return connection_items
 
     @api(version="2.8")
-    def refresh(self, datasource_item: Union[DatasourceItem, str], incremental: bool = False) -> JobItem:
+    def refresh(self, datasource_item: DatasourceItem | str, incremental: bool = False) -> JobItem:
         """
         Refreshes the extract of an existing workbook.
 
@@ -546,47 +546,47 @@ def publish(
         as_job=False,
     ):
         """
-        Publishes a data source to a server, or appends data to an existing
-        data source.
+         Publishes a data source to a server, or appends data to an existing
+         data source.
 
-        This method checks the size of the data source and automatically
-        determines whether the publish the data source in multiple parts or in
-        one operation.
+         This method checks the size of the data source and automatically
+         determines whether the publish the data source in multiple parts or in
+         one operation.
 
-        REST API: https://help.tableau.com/current/api/rest_api/en-us/REST/rest_api_ref_data_sources.htm#publish_data_source
+         REST API: https://help.tableau.com/current/api/rest_api/en-us/REST/rest_api_ref_data_sources.htm#publish_data_source
 
-        Parameters
-        ----------
-        datasource_item : DatasourceItem
-            The datasource item to publish. The fields for name and project_id
-            are required.
+         Parameters
+         ----------
+         datasource_item : DatasourceItem
+             The datasource item to publish. The fields for name and project_id
+             are required.
 
-        file : PathOrFileR
-            The file path or file object to publish.
+         file : PathOrFileR
+             The file path or file object to publish.
 
-        mode : str
-            Specifies whether you are publishing a new datasource (CreateNew),
-            overwriting an existing datasource (Overwrite), or add to an
-            existing datasource (Append). You can also use the publish mode
-            attributes, for example: TSC.Server.PublishMode.Overwrite.
+         mode : str
+             Specifies whether you are publishing a new datasource (CreateNew),
+             overwriting an existing datasource (Overwrite), or add to an
+             existing datasource (Append). You can also use the publish mode
+             attributes, for example: TSC.Server.PublishMode.Overwrite.
 
-        connection_credentials : Optional[ConnectionCredentials]
-            The connection credentials to use when publishing the datasource.
-            Mutually exclusive with the connections parameter.
+         connection_credentials : Optional[ConnectionCredentials]
+             The connection credentials to use when publishing the datasource.
+             Mutually exclusive with the connections parameter.
 
-        connections : Optional[Sequence[ConnectionItem]]
-            The connections to use when publishing the datasource. Mutually
-            exclusive with the connection_credentials parameter.
+         connections : Optional[Sequence[ConnectionItem]]
+             The connections to use when publishing the datasource. Mutually
+             exclusive with the connection_credentials parameter.
 
-        as_job : bool, default False
-            If True, the publish operation is asynchronous and returns a job
-            item. If False, the publish operation is synchronous and returns a
-            datasource item.
+         as_job : bool, default False
+             If True, the publish operation is asynchronous and returns a job
+             item. If False, the publish operation is synchronous and returns a
+             datasource item.
 
-        Returns
-        -------
-        Union[DatasourceItem, JobItem]
-            The datasource item or job item.
+         Returns
+         -------
+        DatasourceItem | JobItem
+             The datasource item or job item.
 
         """
         if isinstance(file, (os.PathLike, str)):
@@ -687,7 +687,7 @@ def publish(
     @api(version="3.13")
     def update_hyper_data(
         self,
-        datasource_or_connection_item: Union[DatasourceItem, ConnectionItem, str],
+        datasource_or_connection_item: DatasourceItem | ConnectionItem | str,
         *,
         request_id: str,
         actions: Sequence[HyperAction],
@@ -709,7 +709,7 @@ def update_hyper_data(
 
         Parameters
         ----------
-        datasource_or_connection_item : Union[DatasourceItem, ConnectionItem, str]
+        datasource_or_connection_item : DatasourceItem | ConnectionItem | str
             The datasource item, connection item, or datasource ID. Either a
             DataSourceItem or a ConnectionItem. If the datasource only contains
             a single connection, the DataSourceItem is sufficient to identify
@@ -1103,18 +1103,18 @@ def schedule_extract_refresh(
         return self.parent_srv.schedules.add_to_schedule(schedule_id, datasource=item)
 
     @api(version="1.0")
-    def add_tags(self, item: Union[DatasourceItem, str], tags: Union[Iterable[str], str]) -> set[str]:
+    def add_tags(self, item: DatasourceItem | str, tags: Iterable[str] | str) -> set[str]:
         """
         Adds one or more tags to the specified datasource item.
 
         REST API: https://help.tableau.com/current/api/rest_api/en-us/REST/rest_api_ref_data_sources.htm#add_tags_to_data_source
 
         Parameters
         ----------
-        item : Union[DatasourceItem, str]
+        item : DatasourceItem | str
             The datasource item or ID to add tags to.
 
-        tags : Union[Iterable[str], str]
+        tags : Iterable[str] | str
             The tag or tags to add to the datasource item.
 
         Returns
@@ -1125,18 +1125,18 @@ def add_tags(self, item: Union[DatasourceItem, str], tags: Union[Iterable[str],
         return super().add_tags(item, tags)
 
     @api(version="1.0")
-    def delete_tags(self, item: Union[DatasourceItem, str], tags: Union[Iterable[str], str]) -> None:
+    def delete_tags(self, item: DatasourceItem | str, tags: Iterable[str] | str) -> None:
         """
         Deletes one or more tags from the specified datasource item.
 
         REST API: https://help.tableau.com/current/api/rest_api/en-us/REST/rest_api_ref_data_sources.htm#delete_tag_from_data_source
 
         Parameters
         ----------
-        item : Union[DatasourceItem, str]
+        item : DatasourceItem | str
             The datasource item or ID to delete tags from.
 
-        tags : Union[Iterable[str], str]
+        tags : Iterable[str] | str
             The tag or tags to delete from the datasource item.
 
         Returns

tableauserverclient/server/endpoint/default_permissions_endpoint.py

@@ -14,7 +14,7 @@
 from tableauserverclient.helpers.logging import logger
 
 # these are the only two items that can hold default permissions for another type
-BaseItem = Union[DatabaseItem, ProjectItem]
+BaseItem = DatabaseItem | ProjectItem
 
 
 class _DefaultPermissionsEndpoint(Endpoint):
@@ -39,7 +39,7 @@ def __str__(self):
     __repr__ = __str__
 
     def update_default_permissions(
-        self, resource: BaseItem, permissions: Sequence[PermissionsRule], content_type: Union[Resource, str]
+        self, resource: BaseItem, permissions: Sequence[PermissionsRule], content_type: Resource | str
     ) -> list[PermissionsRule]:
         url = f"{self.owner_baseurl()}/{resource.id}/default-permissions/{plural_type(content_type)}"
         update_req = RequestFactory.Permission.add_req(permissions)
@@ -51,7 +51,7 @@ def update_default_permissions(
         return permissions
 
     def delete_default_permission(
-        self, resource: BaseItem, rule: PermissionsRule, content_type: Union[Resource, str]
+        self, resource: BaseItem, rule: PermissionsRule, content_type: Resource | str
     ) -> None:
         for capability, mode in rule.capabilities.items():
             # Made readability better but line is too long, will make this look better
@@ -74,7 +74,7 @@ def delete_default_permission(
 
         logger.info(f"Deleted permission for {rule.grantee.tag_name} {rule.grantee.id} item {resource.id}")
 
-    def populate_default_permissions(self, item: BaseItem, content_type: Union[Resource, str]) -> None:
+    def populate_default_permissions(self, item: BaseItem, content_type: Resource | str) -> None:
         if not item.id:
             error = "Server item is missing ID. Item must be retrieved from server first."
             raise MissingRequiredFieldError(error)
@@ -86,7 +86,7 @@ def permission_fetcher() -> list[PermissionsRule]:
         logger.info(f"Populated default {content_type} permissions for item (ID: {item.id})")
 
     def _get_default_permissions(
-        self, item: BaseItem, content_type: Union[Resource, str], req_options: Optional["RequestOptions"] = None
+        self, item: BaseItem, content_type: Resource | str, req_options: "RequestOptions | None" = None
     ) -> list[PermissionsRule]:
         url = f"{self.owner_baseurl()}/{item.id}/default-permissions/{plural_type(content_type)}"
         server_response = self.get_request(url, req_options)

tableauserverclient/server/endpoint/endpoint.py

@@ -87,7 +87,7 @@ def set_user_agent(parameters):
         # return explicitly for testing only
         return parameters
 
-    def _blocking_request(self, method, url, parameters={}) -> Optional[Union["Response", Exception]]:
+    def _blocking_request(self, method, url, parameters={}) -> "Response | Exception | None":
         response = None
         logger.debug(f"[{datetime.timestamp()}] Begin blocking request to {url}")
         try:
@@ -100,7 +100,7 @@ def _blocking_request(self, method, url, parameters={}) -> Optional[Union["Respo
 
     def send_request_while_show_progress_threaded(
         self, method, url, parameters={}, request_timeout=None
-    ) -> Optional[Union["Response", Exception]]:
+    ) -> "Response | Exception | None":
         return self._blocking_request(method, url, parameters)
 
     def _make_request(
@@ -125,7 +125,7 @@ def _make_request(
         # a request can, for stuff like publishing, spin for ages waiting for a response.
         # we need some user-facing activity so they know it's not dead.
         request_timeout = self.parent_srv.http_options.get("timeout") or 0
-        server_response: Optional[Union["Response", Exception]] = self.send_request_while_show_progress_threaded(
+        server_response: "Response | Exception | None" = self.send_request_while_show_progress_threaded(
             method, url, parameters, request_timeout
         )
         logger.debug(f"[{datetime.timestamp()}] Async request returned: received {server_response}")