Make mypy happy

singer_sdk/_python_types.py

@@ -0,0 +1,4 @@
+import os
+from typing import Union
+
+_FilePath = Union[str, os.PathLike]

singer_sdk/configuration/_dict_config.py

@@ -84,13 +84,15 @@ def merge_config_sources(
         A single configuration dictionary.
     """
     config: dict[str, Any] = {}
-    for config_path in inputs:
-        if config_path == "ENV":
+    for config_input in inputs:
+        if config_input == "ENV":
             env_config = parse_environment_config(config_schema, prefix=env_prefix)
             config.update(env_config)
             continue
 
-        if not Path(config_path).is_file():
+        config_path = Path(config_input)
+
+        if not config_path.is_file():
             raise FileNotFoundError(
                 f"Could not locate config file at '{config_path}'."
                 "Please check that the file exists."

singer_sdk/helpers/_util.py

@@ -1,25 +1,28 @@
 """General helper functions, helper classes, and decorators."""
 
 import json
-from pathlib import Path, PurePath
-from typing import Any, Dict, Union, cast
+import os
+from typing import Any, Dict, cast
 
 import pendulum
 
+from singer_sdk._python_types import _FilePath
 
-def read_json_file(path: Union[PurePath, str]) -> Dict[str, Any]:
-    """Read json file, thowing an error if missing."""
+
+def read_json_file(path: _FilePath) -> Dict[str, Any]:
+    """Read json file, throwing an error if missing."""
     if not path:
         raise RuntimeError("Could not open file. Filepath not provided.")
 
-    if not Path(path).exists():
-        msg = f"File at '{path}' was not found."
-        for template in [f"{path}.template"]:
-            if Path(template).exists():
+    if not os.path.exists(path):
+        msg = f"File at '{path!r}' was not found."
+        for template in [f"{path!r}.template"]:
+            if os.path.exists(template):
                 msg += f"\nFor more info, please see the sample template at: {template}"
         raise FileExistsError(msg)
 
-    return cast(dict, json.loads(Path(path).read_text()))
+    with open(path) as f:
+        return cast(dict, json.load(f))
 
 
 def utc_now() -> pendulum.DateTime:

singer_sdk/mapper_base.py

@@ -1,8 +1,7 @@
 """Abstract base class for stream mapper plugins."""
 
 import abc
-from io import FileIO
-from typing import Iterable, List, Tuple, Type
+from typing import IO, Iterable, List, Tuple, Type
 
 import click
 
@@ -107,10 +106,10 @@ def map_batch_message(
     # CLI handler
 
     @classmethod
-    def invoke(
+    def invoke(  # type: ignore[override]
         cls: Type["InlineMapper"],
         config: Tuple[str, ...] = (),
-        file_input: FileIO = None,
+        file_input: IO[str] = None,
     ) -> None:
         """Invoke the mapper.
 
@@ -129,14 +128,14 @@ def invoke(
         )
         mapper.listen(file_input)
 
-    @classproperty
-    def cli(cls) -> click.Command:
+    @classmethod
+    def get_command(cls: Type["InlineMapper"]) -> click.Command:
         """Execute standard CLI handler for inline mappers.
 
         Returns:
             A click.Command object.
         """
-        command = super().cli
+        command = super().get_command()
         command.help = "Execute the Singer mapper."
         command.params.extend(
             [

singer_sdk/plugin_base.py

@@ -5,7 +5,7 @@
 import logging
 import os
 from collections import OrderedDict
-from pathlib import Path, PurePath
+from pathlib import Path
 from types import MappingProxyType
 from typing import (
     Any,
@@ -14,6 +14,7 @@
     List,
     Mapping,
     Optional,
+    Sequence,
     Tuple,
     Type,
     Union,
@@ -24,6 +25,7 @@
 from jsonschema import Draft4Validator, SchemaError, ValidationError
 
 from singer_sdk import metrics
+from singer_sdk._python_types import _FilePath
 from singer_sdk.configuration._dict_config import parse_environment_config
 from singer_sdk.exceptions import ConfigValidationError
 from singer_sdk.helpers._classproperty import classproperty
@@ -80,7 +82,7 @@ def logger(cls) -> logging.Logger:
 
     def __init__(
         self,
-        config: Optional[Union[dict, PurePath, str, List[Union[PurePath, str]]]] = None,
+        config: Optional[Union[dict, _FilePath, Sequence[_FilePath]]] = None,
         parse_env_config: bool = False,
         validate_config: bool = True,
     ) -> None:
@@ -97,7 +99,7 @@ def __init__(
         """
         if not config:
             config_dict = {}
-        elif isinstance(config, str) or isinstance(config, PurePath):
+        elif isinstance(config, (str, bytes, os.PathLike)):
             config_dict = read_json_file(config)
         elif isinstance(config, list):
             config_dict = {}
@@ -401,7 +403,7 @@ def print_about(cls: Type["PluginBase"], format: Optional[str] = None) -> None:
             print(formatted)
 
     @staticmethod
-    def config_from_cli_args(*args: str) -> Tuple[List[str], bool]:
+    def config_from_cli_args(*args: str) -> Tuple[List[Path], bool]:
         """Parse CLI arguments into a config dictionary.
 
         Args:
@@ -435,7 +437,7 @@ def config_from_cli_args(*args: str) -> Tuple[List[str], bool]:
         return config_files, parse_env_config
 
     @abc.abstractclassmethod
-    def invoke(cls: Type["PluginBase"], *args: Any, **kwargs: Any) -> None:
+    def invoke(cls, *args: Any, **kwargs: Any) -> None:
         """Invoke the plugin.
 
         Args:
@@ -482,8 +484,8 @@ def cb_about(
         cls.print_about(format=value)
         ctx.exit()
 
-    @classproperty
-    def cli(cls) -> click.Command:
+    @classmethod
+    def get_command(cls: Type["PluginBase"]) -> click.Command:
         """Handle command line execution.
 
         Returns:
@@ -528,3 +530,13 @@ def cli(cls) -> click.Command:
                 ),
             ],
         )
+
+    @classmethod
+    def cli(cls: Type["PluginBase"]) -> Any:  # noqa: ANN401
+        """Execute standard CLI handler for taps.
+
+        Returns:
+            The return value of the CLI handler.
+        """
+        command = cls.get_command()
+        return command.main()

singer_sdk/tap_base.py

@@ -4,10 +4,11 @@
 import json
 from enum import Enum
 from pathlib import PurePath
-from typing import Any, Dict, List, Optional, Tuple, Type, Union, cast
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Type, Union, cast
 
 import click
 
+from singer_sdk._python_types import _FilePath
 from singer_sdk._singerlib import Catalog
 from singer_sdk.exceptions import MaxRecordsLimitException
 from singer_sdk.helpers import _state
@@ -46,7 +47,7 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
 
     def __init__(
         self,
-        config: Optional[Union[dict, PurePath, str, List[Union[PurePath, str]]]] = None,
+        config: Optional[Union[dict, _FilePath, Sequence[_FilePath]]] = None,
         catalog: Union[PurePath, str, dict, Catalog, None] = None,
         state: Union[PurePath, str, dict, None] = None,
         parse_env_config: bool = False,
@@ -386,7 +387,7 @@ def sync_all(self) -> None:
     # Command Line Execution
 
     @classmethod
-    def invoke(
+    def invoke(  # type: ignore[override]
         cls: Type["Tap"],
         config: Tuple[str, ...] = (),
         state: str = None,
@@ -404,7 +405,7 @@ def invoke(
         config_files, parse_env_config = cls.config_from_cli_args(*config)
 
         tap = cls(
-            config=config_files or None,
+            config=config_files,
             state=state,
             catalog=catalog,
             parse_env_config=parse_env_config,
@@ -471,14 +472,14 @@ def cb_test(
 
         ctx.exit()
 
-    @classproperty
-    def cli(cls) -> click.Command:
+    @classmethod
+    def get_command(cls: Type["Tap"]) -> click.Command:
         """Execute standard CLI handler for taps.
 
         Returns:
             A click.Command object.
         """
-        command = super().cli
+        command = super().get_command()
         command.help = "Execute the Singer tap."
         command.params.extend(
             [
@@ -526,7 +527,7 @@ class SQLTap(Tap):
 
     def __init__(
         self,
-        config: Optional[Union[dict, PurePath, str, List[Union[PurePath, str]]]] = None,
+        config: Optional[Union[dict, _FilePath, Sequence[_FilePath]]] = None,
         catalog: Union[PurePath, str, dict, None] = None,
         state: Union[PurePath, str, dict, None] = None,
         parse_env_config: bool = False,

singer_sdk/target_base.py

@@ -5,13 +5,12 @@
 import json
 import sys
 import time
-from io import FileIO
-from pathlib import PurePath
-from typing import IO, Counter, Dict, List, Optional, Tuple, Type, Union
+from typing import IO, Counter, Dict, List, Optional, Sequence, Tuple, Type, Union
 
 import click
 from joblib import Parallel, delayed, parallel_backend
 
+from singer_sdk._python_types import _FilePath
 from singer_sdk.exceptions import RecordsWithoutSchemaException
 from singer_sdk.helpers._batch import BaseBatchFileEncoding
 from singer_sdk.helpers._classproperty import classproperty
@@ -43,7 +42,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
 
     def __init__(
         self,
-        config: Optional[Union[dict, PurePath, str, List[Union[PurePath, str]]]] = None,
+        config: Optional[Union[dict, _FilePath, Sequence[_FilePath]]] = None,
         parse_env_config: bool = False,
         validate_config: bool = True,
     ) -> None:
@@ -485,10 +484,10 @@ def _write_state_message(self, state: dict) -> None:
     # CLI handler
 
     @classmethod
-    def invoke(
+    def invoke(  # type: ignore[override]
         cls: Type["Target"],
         config: Tuple[str, ...] = (),
-        file_input: FileIO = None,
+        file_input: IO[str] = None,
     ) -> None:
         """Invoke the target.
 
@@ -507,14 +506,14 @@ def invoke(
         )
         target.listen(file_input)
 
-    @classproperty
-    def cli(cls) -> click.Command:
+    @classmethod
+    def get_command(cls: Type["Target"]) -> click.Command:
         """Execute standard CLI handler for taps.
 
         Returns:
             A click.Command object.
         """
-        command = super().cli
+        command = super().get_command()
         command.help = "Execute the Singer target."
         command.params.extend(
             [

tests/core/test_target_input.py

@@ -39,7 +39,7 @@ def config_file_path(target):
 
 def test_input_arg(cli_runner, config_file_path, target):
     result = cli_runner.invoke(
-        target.cli,
+        target.get_command(),
         [
             "--config",
             config_file_path,