diff --git a/my/core/__init__.py b/my/core/__init__.py index c79e36e..0ba8bda 100644 --- a/my/core/__init__.py +++ b/my/core/__init__.py @@ -1,10 +1,10 @@ # this file only keeps the most common & critical types/utility functions from .common import get_files, PathIsh, Paths from .common import Json -from .common import warn_if_empty from .common import stat, Stats from .common import datetime_naive, datetime_aware from .compat import assert_never +from .utils.itertools import warn_if_empty from .cfg import make_config from .error import Res, unwrap diff --git a/my/core/common.py b/my/core/common.py index 389dedc..f84a395 100644 --- a/my/core/common.py +++ b/my/core/common.py @@ -185,64 +185,6 @@ def get_valid_filename(s: str) -> str: return re.sub(r'(?u)[^-\w.]', '', s) -from typing import Generic, Sized, Callable - - -# X = TypeVar('X') -def _warn_iterator(it, f: Any=None): - emitted = False - for i in it: - yield i - emitted = True - if not emitted: - warnings.warn(f"Function {f} didn't emit any data, make sure your config paths are correct") - - -# TODO ugh, so I want to express something like: -# X = TypeVar('X') -# C = TypeVar('C', bound=Iterable[X]) -# _warn_iterable(it: C) -> C -# but apparently I can't??? ugh. -# https://github.com/python/typing/issues/548 -# I guess for now overloads are fine... - -from typing import overload -X = TypeVar('X') -@overload -def _warn_iterable(it: List[X] , f: Any=None) -> List[X] : ... -@overload -def _warn_iterable(it: Iterable[X], f: Any=None) -> Iterable[X]: ... -def _warn_iterable(it, f=None): - if isinstance(it, Sized): - sz = len(it) - if sz == 0: - warnings.warn(f"Function {f} returned empty container, make sure your config paths are correct") - return it - else: - return _warn_iterator(it, f=f) - - -# ok, this seems to work... -# https://github.com/python/mypy/issues/1927#issue-167100413 -FL = TypeVar('FL', bound=Callable[..., List]) -FI = TypeVar('FI', bound=Callable[..., Iterable]) - -@overload -def warn_if_empty(f: FL) -> FL: ... -@overload -def warn_if_empty(f: FI) -> FI: ... - - -def warn_if_empty(f): - from functools import wraps - - @wraps(f) - def wrapped(*args, **kwargs): - res = f(*args, **kwargs) - return _warn_iterable(res, f=f) - return wrapped - - # global state that turns on/off quick stats # can use the 'quick_stats' contextmanager # to enable/disable this in cli so that module 'stats' @@ -586,6 +528,12 @@ if not TYPE_CHECKING: return UI.listify(*args, **kwargs) + @deprecated('use my.core.warn_if_empty instead') + def warn_if_empty(*args, **kwargs): + from .utils import itertools as UI + + return UI.listify(*args, **kwargs) + # todo wrap these in deprecated decorator as well? from .cachew import mcachew # noqa: F401 diff --git a/my/core/tests/test_common.py b/my/core/tests/test_common.py deleted file mode 100644 index a2019e4..0000000 --- a/my/core/tests/test_common.py +++ /dev/null @@ -1,54 +0,0 @@ -from typing import Iterable, List -import warnings - -from ..common import ( - warn_if_empty, - _warn_iterable, -) - - -def test_warn_if_empty() -> None: - @warn_if_empty - def nonempty() -> Iterable[str]: - yield 'a' - yield 'aba' - - @warn_if_empty - def empty() -> List[int]: - return [] - - # should be rejected by mypy! - # todo how to actually test it? - # @warn_if_empty - # def baad() -> float: - # return 0.00 - - # reveal_type(nonempty) - # reveal_type(empty) - - with warnings.catch_warnings(record=True) as w: - assert list(nonempty()) == ['a', 'aba'] - assert len(w) == 0 - - eee = empty() - assert eee == [] - assert len(w) == 1 - - -def test_warn_iterable() -> None: - i1: List[str] = ['a', 'b'] - i2: Iterable[int] = iter([1, 2, 3]) - # reveal_type(i1) - # reveal_type(i2) - x1 = _warn_iterable(i1) - x2 = _warn_iterable(i2) - # vvvv this should be flagged by mypy - # _warn_iterable(123) - # reveal_type(x1) - # reveal_type(x2) - with warnings.catch_warnings(record=True) as w: - assert x1 is i1 # should be unchanged! - assert len(w) == 0 - - assert list(x2) == [1, 2, 3] - assert len(w) == 0 diff --git a/my/core/utils/itertools.py b/my/core/utils/itertools.py index 7046acf..3997310 100644 --- a/my/core/utils/itertools.py +++ b/my/core/utils/itertools.py @@ -4,7 +4,8 @@ Various helpers/transforms of iterators Ideally this should be as small as possible and we should rely on stdlib itertools or more_itertools """ -from typing import Callable, Dict, Iterable, Iterator, TypeVar, List, cast, TYPE_CHECKING +from typing import Callable, Dict, Iterable, Iterator, Sized, TypeVar, List, cast, TYPE_CHECKING +import warnings from ..compat import ParamSpec @@ -115,3 +116,105 @@ def test_listify() -> None: res = it() assert_type(res, List[int]) assert res == [1, 2] + + +@decorator +def _warn_if_empty(func, *args, **kwargs): + # so there is a more_itertools.peekable which could work nicely for these purposes + # the downside is that it would start advancing the generator right after it's created + # , which can be somewhat confusing + iterable = func(*args, **kwargs) + + if isinstance(iterable, Sized): + sz = len(iterable) + if sz == 0: + # todo use hpi warnings here? + warnings.warn(f"Function {func} returned empty container, make sure your config paths are correct") + return iterable + else: # must be an iterator + + def wit(): + empty = True + for i in iterable: + yield i + empty = False + if empty: + warnings.warn(f"Function {func} didn't emit any data, make sure your config paths are correct") + + return wit() + + +if TYPE_CHECKING: + FF = TypeVar('FF', bound=Callable[..., Iterable]) + + def warn_if_empty(f: FF) -> FF: ... + +else: + warn_if_empty = _warn_if_empty + + +def test_warn_if_empty_iterator() -> None: + from ..compat import assert_type + + @warn_if_empty + def nonempty() -> Iterator[str]: + yield 'a' + yield 'aba' + + with warnings.catch_warnings(record=True) as w: + res1 = nonempty() + assert len(w) == 0 # warning isn't emitted until iterator is consumed + assert_type(res1, Iterator[str]) + # assert isinstance(res1, generator) # FIXME ??? how + assert list(res1) == ['a', 'aba'] + assert len(w) == 0 + + @warn_if_empty + def empty() -> Iterator[int]: + yield from [] + + with warnings.catch_warnings(record=True) as w: + res2 = empty() + assert len(w) == 0 # warning isn't emitted until iterator is consumed + assert_type(res2, Iterator[int]) + # assert isinstance(res1, generator) # FIXME ??? how + assert list(res2) == [] + assert len(w) == 1 + + +def test_warn_if_empty_list() -> None: + from ..compat import assert_type + + ll = [1, 2, 3] + + @warn_if_empty + def nonempty() -> List[int]: + return ll + + + with warnings.catch_warnings(record=True) as w: + res1 = nonempty() + assert len(w) == 0 + assert_type(res1, List[int]) + assert isinstance(res1, list) + assert res1 is ll # object should be unchanged! + + + @warn_if_empty + def empty() -> List[str]: + return [] + + + with warnings.catch_warnings(record=True) as w: + res2 = empty() + assert len(w) == 1 + assert_type(res2, List[str]) + assert isinstance(res2, list) + assert res2 == [] + + +def test_warn_if_empty_unsupported() -> None: + # these should be rejected by mypy! (will show "unused type: ignore" if we break it) + @warn_if_empty # type: ignore[type-var] + def bad_return_type() -> float: + return 0.00 diff --git a/my/ip/all.py b/my/ip/all.py index f4cdb37..46c1fec 100644 --- a/my/ip/all.py +++ b/my/ip/all.py @@ -11,7 +11,7 @@ REQUIRES = ["git+https://github.com/seanbreckenridge/ipgeocache"] from typing import Iterator -from my.core.common import Stats, warn_if_empty +from my.core import Stats, warn_if_empty from my.ip.common import IP