Standardize pipeline factory constructor + related changes.

* Run feeds hooks in PipelineBase.
* Make Pipeline entirely responsible for its use of parallel map().

Author: lemon24

src/reader/_update/__init__.py

@@ -10,7 +10,6 @@
 from itertools import chain
 from typing import Any
 from typing import NamedTuple
-from typing import TYPE_CHECKING
 
 from .._parser import EntryPair
 from .._parser import EntryPairBase
@@ -23,16 +22,13 @@
 from .._types import FeedForUpdate
 from .._types import FeedToUpdate
 from .._types import FeedUpdateIntent
-from .._utils import MapFunction
+from .._utils import make_pool_map
 from .._utils import PrefixLogger
 from ..exceptions import ParseError
 from ..types import ExceptionInfo
 from ..types import UpdateConfig
 from .base import PipelineBase
 
-if TYPE_CHECKING:  # pragma: no cover
-    from ..core import Reader
-
 log = logging.getLogger("reader")
 
 HASH_CHANGED_LIMIT = 24
@@ -53,6 +49,19 @@ class Decider:
 
     old_feed: FeedForUpdate
     now: datetime
+
+    # global now, is used as first_updated_epoch for all new entries,
+    # so that the subset of new entries from an update appears before
+    # all others and the entries in it are sorted by published/updated;
+    # if we used last_updated (now) for this, they would be sorted
+    # by feed order first (due to now increasing for each feed).
+    #
+    # A side effect of relying first_updated_epoch for ordering is that
+    # for the second of two new feeds updated in the same update_feeds()
+    # call, first_updated_epoch != last_updated.
+    #
+    # However, added == last_updated for the first update.
+    #
     global_now: datetime
     config: UpdateConfig
     log: Any = log
@@ -325,7 +334,7 @@ def next_update_after(now: datetime, interval: int, jitter: float = 0) -> dateti
     return rv
 
 
-@dataclass(frozen=True)
+@dataclass
 class Pipeline(PipelineBase[FeedData, EntryData, FeedUpdateIntent, EntryUpdateIntent]):
     """Update multiple feeds.
 
@@ -349,23 +358,6 @@ class Pipeline(PipelineBase[FeedData, EntryData, FeedUpdateIntent, EntryUpdateIn
 
     """
 
-    reader: Reader
-
-    # global now, is used as first_updated_epoch for all new entries,
-    # so that the subset of new entries from an update appears before
-    # all others and the entries in it are sorted by published/updated;
-    # if we used last_updated (now) for this, they would be sorted
-    # by feed order first (due to now increasing for each feed).
-    #
-    # A side effect of relying first_updated_epoch for ordering is that
-    # for the second of two new feeds updated in the same update_feeds()
-    # call, first_updated_epoch != last_updated.
-    #
-    # However, added == last_updated for the first update.
-    #
-    global_now: datetime
-
-    map: MapFunction[Any, Any]
     decider = Decider
 
     def parse_feeds(self, feeds: Iterable[FeedForUpdate]) -> Iterable[ParseResult]:
@@ -388,10 +380,11 @@ def parser_process_feeds_for_update(
                 except ParseError as e:
                     parse_errors.append(ParseResult(feed, e))
 
-        feeds = parser_process_feeds_for_update(feeds)
-        feeds = map(self.decider.process_feed_for_update, feeds)
-        parse_results = self.reader._parser.parallel(feeds, self.map)
-        return chain(parse_results, parse_errors)
+        with make_pool_map(self.workers) as parallel_map:
+            feeds = parser_process_feeds_for_update(feeds)
+            feeds = map(self.decider.process_feed_for_update, feeds)
+            parse_results = self.reader._parser.parallel(feeds, parallel_map)
+            yield from chain(parse_results, parse_errors)
 
     def make_intents(
         self, result: ParseResult, entries: Iterable[EntryPair]
@@ -411,7 +404,7 @@ def make_intents(
 
         return self.decider.make_intents(
             self.reader._now(),
-            self.global_now,
+            self.now,
             config,
             result,
             entries,

src/reader/_update/base.py

@@ -3,9 +3,13 @@
 import logging
 from abc import ABC
 from abc import abstractmethod
+from collections.abc import Callable
 from collections.abc import Iterable
 from collections.abc import Iterator
 from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any  # noqa: F401
 from typing import Generic
 from typing import TYPE_CHECKING
 from typing import TypeVar
@@ -28,6 +32,11 @@
 log = logging.getLogger("reader")
 
 
+PipelineFactory = Callable[
+    ['Reader', datetime, int, bool], 'PipelineBase[Any, Any, Any, Any]'
+]
+
+
 FD = TypeVar('FD')
 ED = TypeVar('ED')
 FI = TypeVar('FI')
@@ -37,6 +46,7 @@
 EntryPair = EntryPairBase[ED]
 
 
+@dataclass
 class PipelineBase(Generic[FD, ED, FI, EI], ABC):
     """Run through high level update phases and call update hooks.
 
@@ -46,6 +56,9 @@ class PipelineBase(Generic[FD, ED, FI, EI], ABC):
     """
 
     reader: Reader
+    now: datetime
+    workers: int
+    call_feeds_hooks: bool
 
     @abstractmethod
     def parse_feeds(
@@ -78,6 +91,10 @@ def get_entry_data(self, intent: EI) -> EntryData:
         """Transform an entry update intent into entry data (for plugins)."""
 
     def update(self, filter: FeedFilter) -> Iterable[UpdateResult]:
+
+        if self.call_feeds_hooks:
+            self.reader._update_hooks.run('before_feeds_update', None)
+
         feeds = self.reader._storage.get_feeds_for_update(filter)
         parse_results = self.parse_feeds(feeds)
         update_results = map(self.process_result, parse_results)
@@ -93,6 +110,12 @@ def update(self, filter: FeedFilter) -> Iterable[UpdateResult]:
 
             yield UpdateResult(url, value)
 
+        if self.call_feeds_hooks:
+            with self.reader._update_hooks.group(
+                "got unexpected after-update hook errors"
+            ) as hook_errors:
+                hook_errors.run('after_feeds_update', None)
+
     def process_result(
         self, result: ParseResult[FD, ED]
     ) -> tuple[str, UpdatedFeed | None | Exception]:
@@ -103,7 +126,7 @@ def process_result(
             feed_intent, entry_intents = self.make_intents(result, entry_pairs)
             entry_intents = list(entry_intents)
 
-            with self.run_hooks(result.feed.url, entry_intents):
+            with self.run_feed_hooks(result.feed.url, entry_intents):
                 self.store_feed(feed_intent, [new for new, _ in entry_intents])
 
         except Exception as e:
@@ -129,7 +152,9 @@ def get_entry_pairs(self, result: ParseResult[FD, ED]) -> Iterable[EntryPair[ED]
         return zip(result.value.entries, entries_for_update, strict=True)
 
     @contextmanager
-    def run_hooks(self, feed: str, entries: Iterable[EntryPair[EI]]) -> Iterator[None]:
+    def run_feed_hooks(
+        self, feed: str, entries: Iterable[EntryPair[EI]]
+    ) -> Iterator[None]:
         hooks = self.reader._update_hooks
 
         hooks.run('before_feed_update', (feed,), feed)

src/reader/_utils.py

@@ -11,8 +11,9 @@
 from collections.abc import Iterable
 from collections.abc import Iterator
 from collections.abc import Sequence
-from contextlib import AbstractContextManager
+from contextlib import AbstractContextManager as CM
 from contextlib import contextmanager
+from contextlib import nullcontext
 from functools import wraps
 from typing import Any
 from typing import cast
@@ -79,14 +80,25 @@ def eager_iterable(it: Iterable[_T]) -> Iterable[_T]:
         return it
 
 
+# if we substitute MapFunction below, mypy complains
+# https://github.com/python/mypy/issues/17551
 MapFunction = Callable[[Callable[[_T], _U], Iterable[_T]], Iterator[_U]]
-MapContextManager = AbstractContextManager[MapFunction[_T, _U]]
 
-# type ignore because of https://github.com/python/mypy/issues/17551
 
+def make_pool_map(
+    workers: int,
+) -> CM[Callable[[Callable[[_T], _U], Iterable[_T]], Iterator[_U]]]:
+    if workers < 1:
+        raise ValueError("workers must be a positive integer")
+    if workers == 1:
+        return nullcontext(map)
+    return _make_pool_map(workers)
 
-@contextmanager  # type: ignore[arg-type]
-def make_pool_map(workers: int) -> Iterator[MapFunction[_T, _U]]:
+
+@contextmanager
+def _make_pool_map(
+    workers: int,
+) -> Iterator[Callable[[Callable[[_T], _U], Iterable[_T]], Iterator[_U]]]:
     # We are using concurrent.futures instead of multiprocessing.dummy
     # because the latter doesn't work on some environments (e.g. AWS Lambda).
     # We are not using executor.map() because it consumes the entire iterable.

src/reader/core.py

@@ -1,14 +1,12 @@
 from __future__ import annotations
 
-import builtins
 import logging
 import numbers
 import warnings
 from collections.abc import Callable
 from collections.abc import Iterable
 from collections.abc import Mapping
 from collections.abc import MutableSequence
-from contextlib import nullcontext
 from datetime import datetime
 from datetime import timezone
 from types import MappingProxyType
@@ -37,8 +35,6 @@
 from ._types import UpdateHooks
 from ._update import Pipeline
 from ._utils import eager_iterable
-from ._utils import make_pool_map
-from ._utils import MapContextManager
 from ._utils import zero_or_one
 from .exceptions import EntryNotFoundError
 from .exceptions import FeedError
@@ -82,6 +78,7 @@
 
 if TYPE_CHECKING:  # pragma: no cover
     from ._parser import Parser
+    from ._update.base import PipelineFactory
 
 
 log = logging.getLogger('reader')
@@ -388,6 +385,7 @@ def __init__(
 
         self._reserved_name_scheme = _reserved_name_scheme
         self._enable_search = _enable_search
+        self._make_pipeline: PipelineFactory = Pipeline
         self._update_hooks = UpdateHooks(self)
 
         #: Override update_feeds(scheduled=...).
@@ -930,7 +928,7 @@ def update_feeds_iter(
         new: bool | None = None,
         scheduled: bool = True,
         workers: int = 1,
-        _call_feeds_update_hooks: bool = True,
+        _call_feeds_hooks: bool = True,
     ) -> Iterable[UpdateResult]:
         r"""Update all or some of the feeds.
 
@@ -1035,24 +1033,8 @@ def update_feeds_iter(
             now, feed, tags, broken, updates_enabled, new, scheduled
         )
 
-        if workers < 1:
-            raise ValueError("workers must be a positive integer")
-
-        make_map: MapContextManager[Any, Any] = (
-            nullcontext(builtins.map) if workers == 1 else make_pool_map(workers)
-        )
-
-        if _call_feeds_update_hooks:
-            self._update_hooks.run('before_feeds_update', None)
-
-        with make_map as map:
-            yield from Pipeline(self, now, map).update(filter)
-
-        if _call_feeds_update_hooks:
-            with self._update_hooks.group(
-                "got unexpected after-update hook errors"
-            ) as hook_errors:
-                hook_errors.run('after_feeds_update', None)
+        pipeline = self._make_pipeline(self, now, workers, _call_feeds_hooks)
+        yield from pipeline.update(filter)
 
     def update_feed(self, feed: FeedInput, /) -> UpdatedFeed | None:
         r"""Update a single feed.
@@ -1102,7 +1084,7 @@ def update_feed(self, feed: FeedInput, /) -> UpdatedFeed | None:
                 feed=feed,
                 updates_enabled=None,
                 scheduled=False,
-                _call_feeds_update_hooks=False,
+                _call_feeds_hooks=False,
             ),
             lambda: FeedNotFoundError(_feed_argument(feed)),
         )

tests/test_reader_update.py

@@ -696,14 +696,14 @@ def target():
 
     from reader._update import Pipeline
 
-    def update_feed(*args, **kwargs):
+    def store_feed(*args, **kwargs):
         monkeypatch.undo()
         block()
         assert reader.get_entry(entry).title == 'two'
-        return Pipeline.update_feed(*args, **kwargs)
+        return Pipeline.store_feed(*args, **kwargs)
 
     # TODO: this would have been easier if Pipeline were a reader attribute
-    monkeypatch.setattr(Pipeline, 'update_feed', update_feed)
+    monkeypatch.setattr(Pipeline, 'store_feed', store_feed)
 
     t = threading.Thread(target=target)
     t.start()
@@ -746,13 +746,13 @@ def target():
 
     from reader._update import Pipeline
 
-    def update_feed(*args, **kwargs):
+    def store_feed(*args, **kwargs):
         monkeypatch.undo()
         block()
-        return Pipeline.update_feed(*args, **kwargs)
+        return Pipeline.store_feed(*args, **kwargs)
 
     # TODO: this would have been easier if Pipeline were a reader attribute
-    monkeypatch.setattr(Pipeline, 'update_feed', update_feed)
+    monkeypatch.setattr(Pipeline, 'store_feed', store_feed)
 
     before_entry = reader.get_entry(entry)
     (before_efu,) = reader._storage.get_entries_for_update([entry.resource_id])