From 0c18a57449a06796bfa337e5118de3a6c908785a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edgar=20Rami=CC=81rez=20Mondrago=CC=81n?= Date: Wed, 31 Aug 2022 01:04:14 -0500 Subject: [PATCH] Make mypy happy --- singer_sdk/_python_types.py | 4 ++++ singer_sdk/configuration/_dict_config.py | 8 +++++--- singer_sdk/helpers/_util.py | 21 +++++++++++-------- singer_sdk/mapper_base.py | 13 ++++++------ singer_sdk/plugin_base.py | 26 +++++++++++++++++------- singer_sdk/tap_base.py | 17 ++++++++-------- singer_sdk/target_base.py | 17 ++++++++-------- tests/core/test_target_input.py | 2 +- 8 files changed, 64 insertions(+), 44 deletions(-) create mode 100644 singer_sdk/_python_types.py diff --git a/singer_sdk/_python_types.py b/singer_sdk/_python_types.py new file mode 100644 index 0000000000..75c5ef2747 --- /dev/null +++ b/singer_sdk/_python_types.py @@ -0,0 +1,4 @@ +import os +from typing import Union + +_FilePath = Union[str, os.PathLike] diff --git a/singer_sdk/configuration/_dict_config.py b/singer_sdk/configuration/_dict_config.py index f5061f07a0..03bbd9f1d3 100644 --- a/singer_sdk/configuration/_dict_config.py +++ b/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." diff --git a/singer_sdk/helpers/_util.py b/singer_sdk/helpers/_util.py index 67bc8e8162..7733d72b37 100644 --- a/singer_sdk/helpers/_util.py +++ b/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: diff --git a/singer_sdk/mapper_base.py b/singer_sdk/mapper_base.py index d932ad5fcd..6dff4ded80 100644 --- a/singer_sdk/mapper_base.py +++ b/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 import singer @@ -90,10 +89,10 @@ def map_activate_version_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. @@ -112,14 +111,14 @@ def invoke( ) mapper.listen(file_input) - @classproperty - def cli(cls) -> click.Command: + @classmethod + def get_cli(cls: Type["InlineMapper"]) -> click.Command: """Execute standard CLI handler for inline mappers. Returns: A click.Command object. """ - command = super().cli + command = super().get_cli() command.help = "Execute the Singer mapper." command.params.extend( [ diff --git a/singer_sdk/plugin_base.py b/singer_sdk/plugin_base.py index 68bee956d0..da33fd259c 100644 --- a/singer_sdk/plugin_base.py +++ b/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, @@ -23,6 +24,7 @@ import click from jsonschema import Draft4Validator, SchemaError, ValidationError +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 @@ -79,7 +81,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: @@ -96,7 +98,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 = {} @@ -398,7 +400,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: @@ -432,7 +434,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: @@ -479,8 +481,8 @@ def cb_about( cls.print_about(format=value) ctx.exit() - @classproperty - def cli(cls) -> click.Command: + @classmethod + def get_cli(cls: Type["PluginBase"]) -> click.Command: """Handle command line execution. Returns: @@ -525,3 +527,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. + """ + app = cls.get_cli() + return app() diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py index 38f63039f4..350a672eda 100644 --- a/singer_sdk/tap_base.py +++ b/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.exceptions import MaxRecordsLimitException from singer_sdk.helpers import _state from singer_sdk.helpers._classproperty import classproperty @@ -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_cli(cls: Type["Tap"]) -> click.Command: """Execute standard CLI handler for taps. Returns: A click.Command object. """ - command = super().cli + command = super().get_cli() 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, diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py index c92e08f06b..d44a317a7a 100644 --- a/singer_sdk/target_base.py +++ b/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 RecordsWitoutSchemaException from singer_sdk.helpers._classproperty import classproperty from singer_sdk.helpers._compat import final @@ -42,7 +41,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: @@ -468,10 +467,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. @@ -490,14 +489,14 @@ def invoke( ) target.listen(file_input) - @classproperty - def cli(cls) -> click.Command: + @classmethod + def get_cli(cls: Type["Target"]) -> click.Command: """Execute standard CLI handler for taps. Returns: A click.Command object. """ - command = super().cli + command = super().get_cli() command.help = "Execute the Singer target." command.params.extend( [ diff --git a/tests/core/test_target_input.py b/tests/core/test_target_input.py index f4e91133d8..7337ac2d18 100644 --- a/tests/core/test_target_input.py +++ b/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_cli(), [ "--config", config_file_path,