diff --git a/my/core/__init__.py b/my/core/__init__.py index 45152355..678df850 100644 --- a/my/core/__init__.py +++ b/my/core/__init__.py @@ -1,4 +1,5 @@ # this file only keeps the most common & critical types/utility functions from .common import PathIsh, Paths, Json from .common import get_files, LazyLogger +from .common import warn_if_empty from .cfg import make_config diff --git a/my/core/common.py b/my/core/common.py index cfadc041..64e7b233 100644 --- a/my/core/common.py +++ b/my/core/common.py @@ -3,7 +3,7 @@ from datetime import datetime import functools import types -from typing import Union, Callable, Dict, Iterable, TypeVar, Sequence, List, Optional, Any, cast, Tuple +from typing import Union, Callable, Dict, Iterable, TypeVar, Sequence, List, Optional, Any, cast, Tuple, TYPE_CHECKING import warnings # some helper functions @@ -130,6 +130,11 @@ def get_files(pp: Paths, glob: str=DEFAULT_GLOB, sort: bool=True) -> Tuple[Path, else: sources.extend(map(Path, pp)) + def caller() -> str: + import traceback + # TODO ugh. very flaky... -3 because [, get_files(), ] + return traceback.extract_stack()[-3].filename + paths: List[Path] = [] for src in sources: if src.parts[0] == '~': @@ -141,7 +146,7 @@ def get_files(pp: Paths, glob: str=DEFAULT_GLOB, sort: bool=True) -> Tuple[Path, ss = str(src) if '*' in ss: if glob != DEFAULT_GLOB: - warnings.warn(f"Treating {ss} as glob path. Explicit glob={glob} argument is ignored!") + warnings.warn(f"{caller()}: treating {ss} as glob path. Explicit glob={glob} argument is ignored!") paths.extend(map(Path, do_glob(ss))) else: if not src.is_file(): @@ -154,14 +159,15 @@ def get_files(pp: Paths, glob: str=DEFAULT_GLOB, sort: bool=True) -> Tuple[Path, if len(paths) == 0: # todo make it conditionally defensive based on some global settings - # todo stacktrace? - warnings.warn(f'No paths were matched against {paths}. This might result in missing data.') + # TODO not sure about using warnings module for this + import traceback + warnings.warn(f'{caller()}: no paths were matched against {paths}. This might result in missing data.') + traceback.print_stack() return tuple(paths) # TODO annotate it, perhaps use 'dependent' type (for @doublewrap stuff) -from typing import TYPE_CHECKING if TYPE_CHECKING: from typing import Callable, TypeVar from typing_extensions import Protocol @@ -269,3 +275,60 @@ def isoparse(s: str) -> tzdatetime: def get_valid_filename(s: str) -> str: s = str(s).strip().replace(' ', '_') 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 # type: ignore diff --git a/my/rss/all.py b/my/rss/all.py index 90f5efac..61f9fab1 100644 --- a/my/rss/all.py +++ b/my/rss/all.py @@ -1,11 +1,16 @@ ''' Unified RSS data, merged from different services I used historically ''' +# NOTE: you can comment out the sources you're not using +from . import feedbin, feedly + from typing import Iterable from .common import Subscription, compute_subscriptions def subscriptions() -> Iterable[Subscription]: - from . import feedbin, feedly # TODO google reader? - yield from compute_subscriptions(feedbin.states(), feedly.states()) + yield from compute_subscriptions( + feedbin.states(), + feedly .states(), + ) diff --git a/my/rss/common.py b/my/rss/common.py index 3dc761c2..9aa5ed81 100644 --- a/my/rss/common.py +++ b/my/rss/common.py @@ -17,6 +17,8 @@ class Subscription(NamedTuple): SubscriptionState = Tuple[datetime, Sequence[Subscription]] +from ..core import warn_if_empty +@warn_if_empty def compute_subscriptions(*sources: Iterable[SubscriptionState]) -> List[Subscription]: """ Keeps track of everything I ever subscribed to. @@ -34,6 +36,9 @@ def compute_subscriptions(*sources: Iterable[SubscriptionState]) -> List[Subscri if feed.url not in by_url: by_url[feed.url] = feed + if len(states) == 0: + return [] + _, last_state = max(states, key=lambda x: x[0]) last_urls = {f.url for f in last_state} diff --git a/my/twitter/all.py b/my/twitter/all.py index acb59a28..5c8103cc 100644 --- a/my/twitter/all.py +++ b/my/twitter/all.py @@ -3,11 +3,9 @@ """ # NOTE: you can comment out the sources you don't need - - from . import twint, archive -from .common import merge_tweets +from .common import merge_tweets def tweets(): yield from merge_tweets( @@ -15,6 +13,7 @@ def tweets(): archive.tweets(), ) +from .common import merge_tweets def likes(): yield from merge_tweets( diff --git a/my/twitter/common.py b/my/twitter/common.py index 1bf36f07..ecfaea37 100644 --- a/my/twitter/common.py +++ b/my/twitter/common.py @@ -2,7 +2,9 @@ from more_itertools import unique_everseen +from ..core import warn_if_empty +@warn_if_empty def merge_tweets(*sources): yield from unique_everseen( chain(*sources), diff --git a/tests/misc.py b/tests/misc.py index 40d63a41..89308517 100644 --- a/tests/misc.py +++ b/tests/misc.py @@ -48,3 +48,54 @@ def prepare(tmp_path: Path): # meh from my.core.error import test_sort_res_by + + +from typing import Iterable, List +import warnings +from my.core import warn_if_empty +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: + from my.core.common import _warn_iterable + 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