riteshghorse commented on code in PR #30307:
URL: https://github.com/apache/beam/pull/30307#discussion_r1491805450
##########
sdks/python/apache_beam/io/requestresponse.py:
##########
@@ -251,75 +285,14 @@ def __init__(
self.delay_secs = delay_secs
-class RequestResponseIO(beam.PTransform[beam.PCollection[RequestT],
- beam.PCollection[ResponseT]]):
- """A :class:`RequestResponseIO` transform to read and write to APIs.
-
- Processes an input :class:`~apache_beam.pvalue.PCollection` of requests
- by making a call to the API as defined in :class:`Caller`'s `__call__`
- and returns a :class:`~apache_beam.pvalue.PCollection` of responses.
- """
- def __init__(
- self,
- caller: Caller[RequestT, ResponseT],
- timeout: Optional[float] = DEFAULT_TIMEOUT_SECS,
- should_backoff: Optional[ShouldBackOff] = None,
- repeater: Repeater = ExponentialBackOffRepeater(),
- cache_reader: Optional[CacheReader] = None,
- cache_writer: Optional[CacheWriter] = None,
- throttler: PreCallThrottler = DefaultThrottler(),
- ):
- """
- Instantiates a RequestResponseIO transform.
-
- Args:
- caller (~apache_beam.io.requestresponse.Caller): an implementation of
- `Caller` object that makes call to the API.
- timeout (float): timeout value in seconds to wait for response from API.
- should_backoff (~apache_beam.io.requestresponse.ShouldBackOff):
- (Optional) provides methods for backoff.
- repeater (~apache_beam.io.requestresponse.Repeater): provides method to
- repeat failed requests to API due to service errors. Defaults to
- :class:`apache_beam.io.requestresponse.ExponentialBackOffRepeater` to
- repeat requests with exponential backoff.
- cache_reader (~apache_beam.io.requestresponse.CacheReader): (Optional)
- provides methods to read external cache.
- cache_writer (~apache_beam.io.requestresponse.CacheWriter): (Optional)
- provides methods to write to external cache.
- throttler (~apache_beam.io.requestresponse.PreCallThrottler):
- provides methods to pre-throttle a request. Defaults to
- :class:`apache_beam.io.requestresponse.DefaultThrottler` for
- client-side adaptive throttling using
- :class:`apache_beam.io.components.adaptive_throttler.AdaptiveThrottler`
- """
- self._caller = caller
- self._timeout = timeout
- self._should_backoff = should_backoff
- if repeater:
- self._repeater = repeater
- else:
- self._repeater = NoOpsRepeater()
- self._cache_reader = cache_reader
- self._cache_writer = cache_writer
- self._throttler = throttler
-
- def expand(
- self,
- requests: beam.PCollection[RequestT]) -> beam.PCollection[ResponseT]:
- # TODO(riteshghorse): handle Cache and Throttle PTransforms when available.
- if isinstance(self._throttler, DefaultThrottler):
- return requests | _Call(
- caller=self._caller,
- timeout=self._timeout,
- should_backoff=self._should_backoff,
- repeater=self._repeater,
- throttler=self._throttler)
+class _FilterCacheReadFn(beam.DoFn):
Review Comment:
I would keep it as is here because we care about both hits and misses here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
