Skip to content

Commit

Permalink
Make mypy happy
Browse files Browse the repository at this point in the history
  • Loading branch information
edgarrmondragon committed Sep 1, 2022
1 parent f9d8455 commit b5af06a
Show file tree
Hide file tree
Showing 8 changed files with 64 additions and 44 deletions.
4 changes: 4 additions & 0 deletions singer_sdk/_python_types.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
import os
from typing import Union

_FilePath = Union[str, os.PathLike]
8 changes: 5 additions & 3 deletions singer_sdk/configuration/_dict_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."
Expand Down
21 changes: 12 additions & 9 deletions singer_sdk/helpers/_util.py
Original file line number Diff line number Diff line change
@@ -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:
Expand Down
13 changes: 6 additions & 7 deletions singer_sdk/mapper_base.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -112,14 +111,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(
[
Expand Down
26 changes: 19 additions & 7 deletions singer_sdk/plugin_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -14,6 +14,7 @@
List,
Mapping,
Optional,
Sequence,
Tuple,
Type,
Union,
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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 = {}
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -479,8 +481,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:
Expand Down Expand Up @@ -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.
"""
command = cls.get_command()
return command.main()
17 changes: 9 additions & 8 deletions singer_sdk/tap_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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(
[
Expand Down Expand Up @@ -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,
Expand Down
17 changes: 8 additions & 9 deletions singer_sdk/target_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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._classproperty import classproperty
from singer_sdk.helpers._compat import final
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -469,10 +468,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.
Expand All @@ -491,14 +490,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(
[
Expand Down
2 changes: 1 addition & 1 deletion tests/core/test_target_input.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down

0 comments on commit b5af06a

Please sign in to comment.