apache_beam.io.requestresponseio module

PTransform for reading from and writing to Web APIs.

exception apache_beam.io.requestresponseio.UserCodeExecutionException[source]

Bases: Exception

Base class for errors related to calling Web APIs.

exception apache_beam.io.requestresponseio.UserCodeQuotaException[source]

Bases: apache_beam.io.requestresponseio.UserCodeExecutionException

Extends UserCodeExecutionException to signal specifically that the Web API client encountered a Quota or API overuse related error.

exception apache_beam.io.requestresponseio.UserCodeTimeoutException[source]

Bases: apache_beam.io.requestresponseio.UserCodeExecutionException

Extends UserCodeExecutionException to signal a user code timeout.

class apache_beam.io.requestresponseio.Caller[source]

Bases: contextlib.AbstractContextManager, abc.ABC

Interface for user custom code intended for API calls. For setup and teardown of clients when applicable, implement the __enter__ and __exit__ methods respectively.

class apache_beam.io.requestresponseio.ShouldBackOff[source]

Bases: abc.ABC

ShouldBackOff provides mechanism to apply adaptive throttling.

class apache_beam.io.requestresponseio.Repeater[source]

Bases: abc.ABC

Repeater provides mechanism to repeat requests for a configurable condition.

class apache_beam.io.requestresponseio.CacheReader[source]

Bases: abc.ABC

CacheReader provides mechanism to read from the cache.

class apache_beam.io.requestresponseio.CacheWriter[source]

Bases: abc.ABC

CacheWriter provides mechanism to write to the cache.

class apache_beam.io.requestresponseio.PreCallThrottler[source]

Bases: abc.ABC

PreCallThrottler provides a throttle mechanism before sending request.

class apache_beam.io.requestresponseio.RequestResponseIO(caller: [<class 'apache_beam.io.requestresponseio.Caller'>], timeout: Optional[float] = 30, should_backoff: Optional[apache_beam.io.requestresponseio.ShouldBackOff] = None, repeater: Optional[apache_beam.io.requestresponseio.Repeater] = None, cache_reader: Optional[apache_beam.io.requestresponseio.CacheReader] = None, cache_writer: Optional[apache_beam.io.requestresponseio.CacheWriter] = None, throttler: Optional[apache_beam.io.requestresponseio.PreCallThrottler] = None)[source]

Bases: apache_beam.transforms.ptransform.PTransform

A RequestResponseIO transform to read and write to APIs.

Processes an input PCollection of requests by making a call to the API as defined in Caller’s __call__ and returns a PCollection of responses.

Instantiates a RequestResponseIO transform.

Parameters:
  • caller (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 (ShouldBackOff) – (Optional) provides methods for backoff.
  • repeater (Repeater) – (Optional) provides methods to repeat requests to API.
  • cache_reader (CacheReader) – (Optional) provides methods to read external cache.
  • cache_writer (CacheWriter) – (Optional) provides methods to write to external cache.
  • throttler (PreCallThrottler) – (Optional) provides methods to pre-throttle a request.
expand(requests: apache_beam.pvalue.PCollection[~RequestT][RequestT]) → apache_beam.pvalue.PCollection[~ResponseT][ResponseT][source]