mypy: enable checking untyped code

pyproject.toml

@@ -13,6 +13,7 @@ required-version = 22
 target-version = ['py39', 'py310', 'py311']
 extend-exclude = '^/bin/*'
 [tool.mypy]
+check_untyped_defs = true
 disable_error_code = [
     'misc',
     'import'

tabcmd/commands/auth/login_command.py

@@ -17,9 +17,9 @@ def define_args(parser):
         # just uses global options
         pass
 
-    @staticmethod
-    def run_command(args):
-        logger = log(__class__.__name__, args.logging_level)
+    @classmethod
+    def run_command(cls, args):
+        logger = log(cls.__name__, args.logging_level)
         logger.debug(_("tabcmd.launching"))
         session = Session()
         session.create_session(args, logger)

tabcmd/commands/auth/logout_command.py

@@ -17,9 +17,9 @@ def define_args(parser):
         # has no options
         pass
 
-    @staticmethod
-    def run_command(args):
-        logger = log(__class__.__name__, args.logging_level)
+    @classmethod
+    def run_command(cls, args):
+        logger = log(cls.__name__, args.logging_level)
         logger.debug(_("tabcmd.launching"))
         session = Session()
         session.end_session_and_clear_data()

tabcmd/commands/auth/session.py

@@ -13,7 +13,7 @@
 from tabcmd.execution.localize import _
 from tabcmd.execution.logger_config import log
 
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 
 
 class Session:
@@ -28,31 +28,31 @@ class Session:
     PASSWORD_CRED_TYPE = "password"
 
     def __init__(self):
-        self.username = None
+        self.username: Optional[str] = None
         # we don't store the password
-        self.user_id = None
-        self.auth_token = None
-        self.token_name = None
-        self.token_value = None
-        self.password_file = None
-        self.token_file = None
-        self.site_name = None  # The site name, e.g 'alpodev'
-        self.site_id = None  # The site id, e.g 'abcd-1234-1234-1244-1234'
-        self.server_url = None
-        self.last_command = None  # for when we have to renew the session then re-try
-        self.last_login_using = None
-
-        self.no_prompt = False
-        self.certificate = None
-        self.no_certcheck = False
-        self.no_proxy = False
-        self.proxy = None
-        self.timeout = None
-
-        self.logging_level = "info"
+        self.user_id: Optional[str] = None
+        self.auth_token: Optional[str] = None
+        self.token_name: Optional[str] = None
+        self.token_value: Optional[str] = None
+        self.password_file: Optional[str] = None
+        self.token_file: Optional[str] = None
+        self.site_name: Optional[str] = None  # The site name, e.g 'alpodev'
+        self.site_id: Optional[str] = None  # The site id, e.g 'abcd-1234-1234-1244-1234'
+        self.server_url: Optional[str] = None
+        self.last_command: Optional[str] = None  # for when we have to renew the session then re-try
+        self.last_login_using: Optional[str] = None
+
+        self.no_prompt: bool = False
+        self.certificate: Optional[str] = None
+        self.no_certcheck: bool = False
+        self.no_proxy: bool = False
+        self.proxy: Optional[str] = None
+        self.timeout: Optional[int] = None
+
+        self.logging_level: str = "info"
         self.logger = log(__name__, self.logging_level)  # instantiate here mostly for tests
         self._read_from_json()
-        self.tableau_server = None  # this one is an object that doesn't get persisted in the file
+        self.tableau_server: Optional[TSC.Server] = None  # this one is an object that doesn't get persisted in the file
 
     # called before we connect to the server
     # generally, we don't want to overwrite stored data with nulls
@@ -201,6 +201,10 @@ def _open_connection_with_opts(self) -> TSC.Server:
         return tableau_server
 
     def _verify_server_connection_unauthed(self):
+        if not self.tableau_server:
+            Errors.exit_with_error(self.logger, "No server connection available")
+
+        assert self.tableau_server is not None  # Type hint for mypy
         try:
             self.tableau_server.use_server_version()
         except requests.exceptions.ReadTimeout as timeout_error:
@@ -222,7 +226,12 @@ def _create_new_connection(self) -> TSC.Server:
         try:
             self.tableau_server = self._open_connection_with_opts()
         except Exception as e:
-            Errors.exit_with_error(self.logger, "Failed to connect to server", e)
+            Errors.exit_with_error(self.logger, "Failed to connect to server", exception=e)
+
+        if not self.tableau_server:
+            Errors.exit_with_error(self.logger, "Failed to create server connection")
+
+        assert self.tableau_server is not None  # Type hint for mypy
         return self.tableau_server
 
     def _read_existing_state(self):
@@ -246,7 +255,7 @@ def _print_server_info(self):
     def _validate_existing_signin(self):
         # when do these two messages show up? self.logger.info(_("session.auto_site_login"))
         try:
-            if self.tableau_server and self.tableau_server.is_signed_in():
+            if self.tableau_server and self.tableau_server.is_signed_in() and self.user_id:
                 server_user = self.tableau_server.users.get_by_id(self.user_id).name
                 if not self.username:
                     self.logger.info("Fetched user details from server")
@@ -261,8 +270,12 @@ def _validate_existing_signin(self):
 
     # server connection created, not yet logged in
     def _sign_in(self, tableau_auth) -> TSC.Server:
-        self.logger.debug(_("session.login") + self.server_url)
+        if not self.tableau_server:
+            Errors.exit_with_error(self.logger, "No server connection available for sign in")
+
+        self.logger.debug(_("session.login") + (self.server_url or ""))
         self.logger.debug(_("listsites.output").format("", self.username or self.token_name, self.site_name))
+        assert self.tableau_server is not None  # Type hint for mypy
         try:
             self.tableau_server.auth.sign_in(tableau_auth)  # it's the same call for token or user-pass
         except Exception as e:
@@ -298,7 +311,7 @@ def create_session(self, args, logger):
         self._read_existing_state()
         self._update_session_data(args)
         self.logging_level = args.logging_level or self.logging_level
-        self.logger = logger or log(__class__.__name__, self.logging_level)
+        self.logger = logger or log(self.__class__.__name__, self.logging_level)
 
         credentials = None
         if args.password or args.password_file:
@@ -360,8 +373,8 @@ def _clear_data(self):
         self.tableau_server = None
 
         self.certificate = None
-        self.no_certcheck = None
-        self.no_proxy = None
+        self.no_certcheck = False
+        self.no_proxy = False
         self.proxy = None
         self.timeout = None
 
@@ -440,13 +453,13 @@ def _save_session_to_json(self):
         except Exception as e:
             self._wipe_bad_json(e, "Failed to save session file")
 
-    def _save_file(self, data):
+    def _save_file(self, data: Dict[str, Any]) -> None:
         file_path = self._get_file_path()
         with open(str(file_path), "w") as f:
             json.dump(data, f)
 
-    def _serialize_for_save(self):
-        data = {"tableau_auth": []}
+    def _serialize_for_save(self) -> Dict[str, Any]:
+        data: Dict[str, Any] = {"tableau_auth": []}
         data["tableau_auth"].append(
             {
                 "auth_token": self.auth_token,

tabcmd/commands/datasources_and_workbooks/datasources_and_workbooks_command.py

@@ -5,6 +5,7 @@
 from tabcmd.commands.constants import Errors
 from tabcmd.commands.server import Server
 from tabcmd.execution.localize import _
+RequestOptionsType  = TSC.ExcelRequestOptions | TSC.CSVRequestOptions | TSC.PDFRequestOptions | TSC.ImageRequestOptions
 
 
 class DatasourcesAndWorkbooks(Server):
@@ -13,7 +14,7 @@ class DatasourcesAndWorkbooks(Server):
     """
 
     def __init__(self, args):
-        super().__init__(args)
+        pass
 
     @staticmethod
     def get_view_url_from_names(wb_name, view_name):
@@ -62,7 +63,7 @@ def get_ds_by_content_url(logger, server, datasource_content_url) -> TSC.Datasou
         return matching_datasources[0]
 
     @staticmethod
-    def apply_values_from_url_params(logger, request_options: TSC.RequestOptions, url) -> None:
+    def apply_values_from_url_params(logger, request_options: RequestOptionsType, url) -> None:
         logger.debug(url)
         try:
             if "?" in url:
@@ -86,7 +87,7 @@ def apply_values_from_url_params(logger, request_options: TSC.RequestOptions, ur
 
     # this is called from within from_url_params, for each view_filter value
     @staticmethod
-    def apply_encoded_filter_value(logger, request_options, value):
+    def apply_encoded_filter_value(logger, request_options: RequestOptionsType, value):
         # the REST API doesn't appear to have the option to disambiguate with "Parameters.<fieldname>"
         value = value.replace("Parameters.", "")
         # the filter values received from the url are already url encoded. tsc will encode them again.
@@ -99,7 +100,7 @@ def apply_encoded_filter_value(logger, request_options, value):
     # from apply_options, which expects an un-encoded input,
     # or from apply_url_params via apply_encoded_filter_value which decodes the input
     @staticmethod
-    def apply_filter_value(logger, request_options: TSC.RequestOptions, value: str) -> None:
+    def apply_filter_value(logger, request_options: RequestOptionsType, value: str) -> None:
         logger.debug("handling filter param {}".format(value))
         data_filter = value.split("=")
         # we should export the _DataExportOptions class from tsc
@@ -108,7 +109,7 @@ def apply_filter_value(logger, request_options: TSC.RequestOptions, value: str)
     # this is called from within from_url_params, for each param value
     # expects either ImageRequestOptions or PDFRequestOptions
     @staticmethod
-    def apply_options_in_url(logger, request_options: TSC.RequestOptions, value: str) -> None:
+    def apply_options_in_url(logger, request_options: RequestOptionsType, value: str) -> None:
         logger.debug("handling url option {}".format(value))
         setting = value.split("=")
         if len(setting) != 2:

tabcmd/commands/datasources_and_workbooks/datasources_workbooks_views_url_parser.py

@@ -1,5 +1,6 @@
 import os
 
+import tableauserverclient as TSC
 from uuid import UUID
 
 from tabcmd.commands.constants import Errors
@@ -14,7 +15,7 @@ class DatasourcesWorkbooksAndViewsUrlParser(Server):
     """
 
     def __init__(self, args):
-        super().__init__(args)
+        pass
 
     @staticmethod
     def get_view_url_from_names(wb_name, view_name):
@@ -179,12 +180,12 @@ def get_url_item_and_item_type_from_view_url(logger, url, server):
 
     @staticmethod
     def get_content_and_server_content_type_from_url(logger, server, view_content_url, custom_view_id):
-        item = DatasourcesAndWorkbooks.get_view_by_content_url(logger, server, view_content_url)
+        item: TSC.ViewItem | TSC.CustomViewItem = DatasourcesAndWorkbooks.get_view_by_content_url(logger, server, view_content_url)
         server_content_type = server.views
 
         if custom_view_id:
-            custom_view_item = DatasourcesAndWorkbooks.get_custom_view_by_id(logger, server, custom_view_id)
-            if custom_view_item.view.id != item.id:
+            custom_view_item: TSC.CustomViewItem = DatasourcesAndWorkbooks.get_custom_view_by_id(logger, server, custom_view_id)
+            if not custom_view_item.view or custom_view_item.view.id != item.id:
                 Errors.exit_with_error(logger, "Invalid custom view URL provided")
             server_content_type = server.custom_views
             item = custom_view_item

tabcmd/commands/datasources_and_workbooks/delete_command.py

@@ -27,9 +27,9 @@ def define_args(delete_parser):
         set_project_r_arg(group)
         set_parent_project_arg(group)
 
-    @staticmethod
-    def run_command(args):
-        logger = log(__class__.__name__, args.logging_level)
+    @classmethod
+    def run_command(cls, args):
+        logger = log(cls.__name__, args.logging_level)
         logger.debug(_("tabcmd.launching"))
         session = Session()
         server = session.create_session(args, logger)
@@ -75,4 +75,4 @@ def run_command(args):
                 server.datasources.delete(item_to_delete.id)
             logger.info(_("common.output.succeeded"))
         except Exception as e:
-            Errors.exit_with_error(logger, e)
+            Errors.exit_with_error(logger, exception=e)

tabcmd/commands/datasources_and_workbooks/export_command.py

@@ -13,6 +13,7 @@
 pagesize = TSC.PDFRequestOptions.PageType  # type alias for brevity
 pageorientation = TSC.PDFRequestOptions.Orientation
 imageresolution = TSC.ImageRequestOptions.Resolution
+RequestOptionsType  = TSC.ExcelRequestOptions | TSC.CSVRequestOptions | TSC.PDFRequestOptions | TSC.ImageRequestOptions
 ImageResolutionStandard = "standard"
 
 
@@ -81,9 +82,9 @@ def define_args(export_parser):
     it to a file. This command can also export just the data used for a view_name
     """
 
-    @staticmethod
-    def run_command(args):
-        logger = log(__class__.__name__, args.logging_level)
+    @classmethod
+    def run_command(cls, args):
+        logger = log(cls.__name__, args.logging_level)
         logger.debug(_("tabcmd.launching"))
         session = Session()
         server = session.create_session(args, logger)
@@ -97,7 +98,7 @@ def run_command(args):
         if not view_content_url and not wb_content_url:
             view_example = "/workbook_name/view_name"
             message = "{} [{}]".format(
-                _("export.errors.requires_workbook_view_param").format(__class__.__name__), view_example
+                _("export.errors.requires_workbook_view_param").format("ExportCommand"), view_example
             )
             Errors.exit_with_error(logger, message)
 
@@ -127,7 +128,7 @@ def run_command(args):
                     default_filename = "{}.png".format(export_item.name)
 
         except TSC.ServerResponseError as e:
-            Errors.exit_with_error(logger, _("publish.errors.unexpected_server_response").format(""), e)
+            Errors.exit_with_error(logger, _("publish.errors.unexpected_server_response").format(""), exception=e)
         except Exception as e:
             Errors.exit_with_error(logger, exception=e)
         try:
@@ -138,10 +139,10 @@ def run_command(args):
                 ExportCommand.save_to_file(logger, output, save_name)
 
         except Exception as e:
-            Errors.exit_with_error(logger, "Error saving to file", e)
+            Errors.exit_with_error(logger, "Error saving to file", exception=e)
 
     @staticmethod
-    def apply_filters_from_args(request_options: TSC.RequestOptions, args, logger=None) -> None:
+    def apply_filters_from_args(request_options: RequestOptionsType, args, logger=None) -> None:
         if args.filter:
             logger.debug("filter = {}".format(args.filter))
             params = args.filter.split("&")

tabcmd/commands/datasources_and_workbooks/get_url_command.py

@@ -31,14 +31,14 @@ def define_args(get_url_parser):
         # tabcmd get "/views/Finance/InvestmentGrowth.png?:size=640,480" -f growth.png
         # tabcmd get "/views/Finance/InvestmentGrowth.png?:refresh=yes" -f growth.png
 
-    @staticmethod
-    def run_command(args):
+    @classmethod
+    def run_command(cls, args):
         # A view can be returned in PDF, PNG, or CSV (summary data only) format.
         # A Tableau workbook is returned as a TWB if it connects to a datasource/live connection,
         # or a TWBX if it uses an extract.
         # A Tableau datasource is returned as a TDS if it connects to a live connection,
         # or a TDSX if it uses an extract.
-        logger = log(__class__.__name__, args.logging_level)
+        logger = log(cls.__name__, args.logging_level)
         logger.debug(_("tabcmd.launching"))
         session = Session()
         server = session.create_session(args, logger)

tabcmd/commands/datasources_and_workbooks/publish_command.py

@@ -33,9 +33,9 @@ def define_args(publish_parser):
         set_append_replace_option(group)
         set_parent_project_arg(group)
 
-    @staticmethod
-    def run_command(args):
-        logger = log(__class__.__name__, args.logging_level)
+    @classmethod
+    def run_command(cls, args):
+        logger = log(cls.__name__, args.logging_level)
         logger.debug(_("tabcmd.launching"))
         session = Session()
         server = session.create_session(args, logger)
@@ -50,22 +50,24 @@ def run_command(args):
             project_id = dest_project.id
         except Exception as exc:
             logger.error(exc.__str__())
-            Errors.exit_with_error(logger, _("publish.errors.server_resource_not_found"), exc)
+            Errors.exit_with_error(logger, _("publish.errors.server_resource_not_found"), exception=exc)
 
         publish_mode = PublishCommand.get_publish_mode(args, logger)
 
-        connection = TSC.models.ConnectionItem()
+        connection: TSC.models.ConnectionItem | None = None
         if args.db_username:
+            connection = TSC.models.ConnectionItem()
             connection.connection_credentials = TSC.models.ConnectionCredentials(
                 args.db_username, args.db_password, embed=args.save_db_password
             )
         elif args.oauth_username:
+            connection = TSC.models.ConnectionItem()
             connection.connection_credentials = TSC.models.ConnectionCredentials(
                 args.oauth_username, None, embed=False, oauth=args.save_oauth
             )
         else:
-            logger.debug("No db-username or oauth-username found in command")
             connection = None
+            logger.debug("No db-username or oauth-username found in command")
 
         if connection:
             connections = list()