You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "damondouglas (via GitHub)" <gi...@apache.org> on 2023/11/13 20:27:56 UTC
[PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
damondouglas opened a new pull request, #29421:
URL: https://github.com/apache/beam/pull/29421
This PR addresses #28710 and addresses #28711 by defining Caller and SetupTeardown abstract classes in Python. Implementation tests implement these classes, calling the HTTP supported [.test-infra/mock-apis](https://github.com/apache/beam/tree/master/.test-infra/mock-apis).
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [x] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "damondouglas (via GitHub)" <gi...@apache.org>.
damondouglas commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1391756348
##########
sdks/python/apache_beam/io/requestresponseio_it_test.py:
##########
@@ -0,0 +1,168 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import base64
+import sys
+import unittest
+from dataclasses import dataclass
+from typing import Tuple
+from typing import Union
+
+import urllib3
+
+from apache_beam.io.requestresponseio import Caller
+from apache_beam.io.requestresponseio import UserCodeExecutionException
+from apache_beam.io.requestresponseio import UserCodeQuotaException
+from apache_beam.options.pipeline_options import PipelineOptions
+
+_HTTP_PATH = "/v1/echo"
+_PAYLOAD = base64.b64encode(bytes('payload', 'utf-8'))
+
+
+class EchoITOptions(PipelineOptions):
+ """Shared options for running integration tests on a deployed
+ ``EchoServiceGrpc`` See https://github.com/apache/beam/tree/master/.test
+ -infra/mock-apis#integration for details on how to acquire values
+ required by ``EchoITOptions``.
+ """
+ @classmethod
+ def _add_argparse_args(cls, parser) -> None:
+ parser.add_argument(
+ '--httpEndpointAddress',
+ required=True,
+ dest='http_endpoint_address',
+ help='The HTTP address of the Echo API endpoint; must being with '
+ 'http(s)://')
+ parser.add_argument(
+ '--neverExceedQuotaId',
+ default='echo-should-never-exceed-quota',
+ dest='never_exceed_quota_id',
+ help='The ID for an allocated quota that should never exceed.')
+ parser.add_argument(
+ '--shouldExceedQuotaId',
+ default='echo-should-exceed-quota',
+ dest='should_exceed_quota_id',
+ help='The ID for an allocated quota that should exceed.')
+
+
+# TODO(riteshghorse,damondouglas) replace Echo(Request|Response) with proto
+# generated classes from .test-infra/mock-apis:
+@dataclass
+class EchoRequest:
+ id: str
+ payload: bytes
+
+
+@dataclass
+class EchoResponse:
+ id: str
+ payload: bytes
+
+
+class EchoHTTPCaller(Caller):
+ """Implements ``Caller`` to call the ``EchoServiceGrpc``'s HTTP handler.
+ The purpose of ``EchoHTTPCaller`` is to support integration tests.
+ """
+ def __init__(self, url: str):
+ self.url = url + _HTTP_PATH
+
+ def call(self, request: EchoRequest) -> EchoResponse:
+ """Overrides ``Caller``'s call method invoking the
+ ``EchoServiceGrpc``'s HTTP handler with an ``EchoRequest``, returning
+ either a successful ``EchoResponse`` or throwing either a
+ ``UserCodeExecutionException``, ``UserCodeTimeoutException``,
+ or a ``UserCodeQuotaException``.
+ """
+
+ try:
+ resp = urllib3.request(
+ "POST",
+ self.url,
+ json={
+ "id": request.id, "payload": str(request.payload, 'utf-8')
+ },
+ retries=False)
+
+ if resp.status < 300:
+ resp_body = resp.json()
+ resp_id = resp_body['id']
+ payload = resp_body['payload']
+ return EchoResponse(id=resp_id, payload=bytes(payload, 'utf-8'))
+
+ if resp.status == 429: # Too Many Requests
+ raise UserCodeQuotaException(resp.reason)
+
+ raise UserCodeExecutionException(resp.reason)
+
+ except urllib3.exceptions.HTTPError as e:
+ raise UserCodeExecutionException(e)
+
+
+class EchoHTTPCallerTestIT(unittest.TestCase):
+ options: Union[EchoITOptions, None] = None
+ client: Union[EchoHTTPCaller, None] = None
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.options = EchoITOptions()
Review Comment:
Thank you for reviewing :-). I was confused about the convention for integration tests in the Python SDK. To be on the safe side, should I just remove the requirement flag and check for it when needed in the test?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1395015226
##########
sdks/python/apache_beam/io/requestresponseio.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""``PTransform`` for reading from and writing to Web APIs."""
+import abc
+from typing import TypeVar
+
+RequestT = TypeVar('RequestT')
+ResponseT = TypeVar('ResponseT')
+
+
+class UserCodeExecutionException(Exception):
+ """Base class for errors related to calling Web APIs."""
+
+
+class UserCodeQuotaException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal specifically that
+ the Web API client encountered a Quota or API overuse related error.
+ """
+
+
+class UserCodeTimeoutException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal a user code timeout."""
+
+
+class Caller(metaclass=abc.ABCMeta):
+ """Interfaces user custom code intended for API calls."""
+ @abc.abstractmethod
+ def call(self, request: RequestT) -> ResponseT:
Review Comment:
Not re-using `__call__`?
##########
sdks/python/apache_beam/io/requestresponseio.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""``PTransform`` for reading from and writing to Web APIs."""
+import abc
+from typing import TypeVar
+
+RequestT = TypeVar('RequestT')
+ResponseT = TypeVar('ResponseT')
+
+
+class UserCodeExecutionException(Exception):
+ """Base class for errors related to calling Web APIs."""
+
+
+class UserCodeQuotaException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal specifically that
+ the Web API client encountered a Quota or API overuse related error.
+ """
+
+
+class UserCodeTimeoutException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal a user code timeout."""
+
+
+class Caller(metaclass=abc.ABCMeta):
+ """Interfaces user custom code intended for API calls."""
+ @abc.abstractmethod
+ def call(self, request: RequestT) -> ResponseT:
+ """Calls a Web API with the ``RequestT`` and returns a
+ ``ResponseT``. ``RequestResponseIO`` expects implementations of the
+ call method to throw either a ``UserCodeExecutionException``,
+ ``UserCodeQuotaException``, or ``UserCodeTimeoutException``.
+ """
+ pass
+
+
+class SetupTeardown(metaclass=abc.ABCMeta):
Review Comment:
I wonder if we should go with the contextlib pattern to ensure correctness.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "damondouglas (via GitHub)" <gi...@apache.org>.
damondouglas commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1391766309
##########
sdks/python/apache_beam/io/requestresponseio_it_test.py:
##########
@@ -0,0 +1,168 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import base64
+import sys
+import unittest
+from dataclasses import dataclass
+from typing import Tuple
+from typing import Union
+
+import urllib3
+
+from apache_beam.io.requestresponseio import Caller
+from apache_beam.io.requestresponseio import UserCodeExecutionException
+from apache_beam.io.requestresponseio import UserCodeQuotaException
+from apache_beam.options.pipeline_options import PipelineOptions
+
+_HTTP_PATH = "/v1/echo"
+_PAYLOAD = base64.b64encode(bytes('payload', 'utf-8'))
+
+
+class EchoITOptions(PipelineOptions):
+ """Shared options for running integration tests on a deployed
+ ``EchoServiceGrpc`` See https://github.com/apache/beam/tree/master/.test
+ -infra/mock-apis#integration for details on how to acquire values
+ required by ``EchoITOptions``.
+ """
+ @classmethod
+ def _add_argparse_args(cls, parser) -> None:
+ parser.add_argument(
+ '--httpEndpointAddress',
+ required=True,
+ dest='http_endpoint_address',
+ help='The HTTP address of the Echo API endpoint; must being with '
+ 'http(s)://')
+ parser.add_argument(
+ '--neverExceedQuotaId',
+ default='echo-should-never-exceed-quota',
+ dest='never_exceed_quota_id',
+ help='The ID for an allocated quota that should never exceed.')
+ parser.add_argument(
+ '--shouldExceedQuotaId',
+ default='echo-should-exceed-quota',
+ dest='should_exceed_quota_id',
+ help='The ID for an allocated quota that should exceed.')
+
+
+# TODO(riteshghorse,damondouglas) replace Echo(Request|Response) with proto
+# generated classes from .test-infra/mock-apis:
+@dataclass
+class EchoRequest:
+ id: str
+ payload: bytes
+
+
+@dataclass
+class EchoResponse:
+ id: str
+ payload: bytes
+
+
+class EchoHTTPCaller(Caller):
+ """Implements ``Caller`` to call the ``EchoServiceGrpc``'s HTTP handler.
+ The purpose of ``EchoHTTPCaller`` is to support integration tests.
+ """
+ def __init__(self, url: str):
+ self.url = url + _HTTP_PATH
+
+ def call(self, request: EchoRequest) -> EchoResponse:
+ """Overrides ``Caller``'s call method invoking the
+ ``EchoServiceGrpc``'s HTTP handler with an ``EchoRequest``, returning
+ either a successful ``EchoResponse`` or throwing either a
+ ``UserCodeExecutionException``, ``UserCodeTimeoutException``,
+ or a ``UserCodeQuotaException``.
+ """
+
+ try:
+ resp = urllib3.request(
+ "POST",
+ self.url,
+ json={
+ "id": request.id, "payload": str(request.payload, 'utf-8')
+ },
+ retries=False)
+
+ if resp.status < 300:
+ resp_body = resp.json()
+ resp_id = resp_body['id']
+ payload = resp_body['payload']
+ return EchoResponse(id=resp_id, payload=bytes(payload, 'utf-8'))
+
+ if resp.status == 429: # Too Many Requests
+ raise UserCodeQuotaException(resp.reason)
+
+ raise UserCodeExecutionException(resp.reason)
+
+ except urllib3.exceptions.HTTPError as e:
+ raise UserCodeExecutionException(e)
+
+
+class EchoHTTPCallerTestIT(unittest.TestCase):
+ options: Union[EchoITOptions, None] = None
+ client: Union[EchoHTTPCaller, None] = None
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.options = EchoITOptions()
Review Comment:
Resolving as I removed the `required=True` from the option.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1396362927
##########
sdks/python/apache_beam/io/requestresponseio.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""``PTransform`` for reading from and writing to Web APIs."""
+import abc
+from typing import TypeVar
+
+RequestT = TypeVar('RequestT')
+ResponseT = TypeVar('ResponseT')
+
+
+class UserCodeExecutionException(Exception):
+ """Base class for errors related to calling Web APIs."""
+
+
+class UserCodeQuotaException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal specifically that
+ the Web API client encountered a Quota or API overuse related error.
+ """
+
+
+class UserCodeTimeoutException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal a user code timeout."""
+
+
+class Caller(metaclass=abc.ABCMeta):
+ """Interfaces user custom code intended for API calls."""
+ @abc.abstractmethod
+ def call(self, request: RequestT) -> ResponseT:
Review Comment:
interesting! having `__call__` here would mean we can accept any callables to the RRIO. This could be a big plus for Python RRIO. WDYT @damondouglas ?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29421:
URL: https://github.com/apache/beam/pull/29421#issuecomment-1809041447
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1397530964
##########
sdks/python/apache_beam/io/requestresponseio.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""``PTransform`` for reading from and writing to Web APIs."""
+import abc
+from typing import TypeVar
+
+RequestT = TypeVar('RequestT')
+ResponseT = TypeVar('ResponseT')
+
+
+class UserCodeExecutionException(Exception):
+ """Base class for errors related to calling Web APIs."""
+
+
+class UserCodeQuotaException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal specifically that
+ the Web API client encountered a Quota or API overuse related error.
+ """
+
+
+class UserCodeTimeoutException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal a user code timeout."""
+
+
+class Caller(metaclass=abc.ABCMeta):
+ """Interfaces user custom code intended for API calls."""
+ @abc.abstractmethod
+ def call(self, request: RequestT) -> ResponseT:
+ """Calls a Web API with the ``RequestT`` and returns a
+ ``ResponseT``. ``RequestResponseIO`` expects implementations of the
+ call method to throw either a ``UserCodeExecutionException``,
+ ``UserCodeQuotaException``, or ``UserCodeTimeoutException``.
+ """
+ pass
+
+
+class SetupTeardown(metaclass=abc.ABCMeta):
Review Comment:
contextlib is cool, thank you for introducing that! Didn't know that `with` uses it all the time
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1391737838
##########
sdks/python/apache_beam/io/requestresponseio_it_test.py:
##########
@@ -0,0 +1,168 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import base64
+import sys
+import unittest
+from dataclasses import dataclass
+from typing import Tuple
+from typing import Union
+
+import urllib3
+
+from apache_beam.io.requestresponseio import Caller
+from apache_beam.io.requestresponseio import UserCodeExecutionException
+from apache_beam.io.requestresponseio import UserCodeQuotaException
+from apache_beam.options.pipeline_options import PipelineOptions
+
+_HTTP_PATH = "/v1/echo"
+_PAYLOAD = base64.b64encode(bytes('payload', 'utf-8'))
+
+
+class EchoITOptions(PipelineOptions):
+ """Shared options for running integration tests on a deployed
+ ``EchoServiceGrpc`` See https://github.com/apache/beam/tree/master/.test
+ -infra/mock-apis#integration for details on how to acquire values
+ required by ``EchoITOptions``.
+ """
+ @classmethod
+ def _add_argparse_args(cls, parser) -> None:
+ parser.add_argument(
+ '--httpEndpointAddress',
+ required=True,
+ dest='http_endpoint_address',
+ help='The HTTP address of the Echo API endpoint; must being with '
+ 'http(s)://')
+ parser.add_argument(
+ '--neverExceedQuotaId',
+ default='echo-should-never-exceed-quota',
+ dest='never_exceed_quota_id',
+ help='The ID for an allocated quota that should never exceed.')
+ parser.add_argument(
+ '--shouldExceedQuotaId',
+ default='echo-should-exceed-quota',
+ dest='should_exceed_quota_id',
+ help='The ID for an allocated quota that should exceed.')
+
+
+# TODO(riteshghorse,damondouglas) replace Echo(Request|Response) with proto
+# generated classes from .test-infra/mock-apis:
+@dataclass
+class EchoRequest:
+ id: str
+ payload: bytes
+
+
+@dataclass
+class EchoResponse:
+ id: str
+ payload: bytes
+
+
+class EchoHTTPCaller(Caller):
+ """Implements ``Caller`` to call the ``EchoServiceGrpc``'s HTTP handler.
+ The purpose of ``EchoHTTPCaller`` is to support integration tests.
+ """
+ def __init__(self, url: str):
+ self.url = url + _HTTP_PATH
+
+ def call(self, request: EchoRequest) -> EchoResponse:
+ """Overrides ``Caller``'s call method invoking the
+ ``EchoServiceGrpc``'s HTTP handler with an ``EchoRequest``, returning
+ either a successful ``EchoResponse`` or throwing either a
+ ``UserCodeExecutionException``, ``UserCodeTimeoutException``,
+ or a ``UserCodeQuotaException``.
+ """
+
+ try:
+ resp = urllib3.request(
+ "POST",
+ self.url,
+ json={
+ "id": request.id, "payload": str(request.payload, 'utf-8')
+ },
+ retries=False)
+
+ if resp.status < 300:
+ resp_body = resp.json()
+ resp_id = resp_body['id']
+ payload = resp_body['payload']
+ return EchoResponse(id=resp_id, payload=bytes(payload, 'utf-8'))
+
+ if resp.status == 429: # Too Many Requests
+ raise UserCodeQuotaException(resp.reason)
+
+ raise UserCodeExecutionException(resp.reason)
+
+ except urllib3.exceptions.HTTPError as e:
+ raise UserCodeExecutionException(e)
+
+
+class EchoHTTPCallerTestIT(unittest.TestCase):
+ options: Union[EchoITOptions, None] = None
+ client: Union[EchoHTTPCaller, None] = None
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.options = EchoITOptions()
Review Comment:
since `httpEndpointAddress` is a required argument, this will throw error. This is causing a lot of test suites to fail.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1395014757
##########
sdks/python/apache_beam/io/requestresponseio_it_test.py:
##########
@@ -0,0 +1,172 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import base64
+import sys
+import unittest
+from dataclasses import dataclass
+from typing import Tuple
+from typing import Union
+
+import urllib3
+
+from apache_beam.io.requestresponseio import Caller
+from apache_beam.io.requestresponseio import UserCodeExecutionException
+from apache_beam.io.requestresponseio import UserCodeQuotaException
+from apache_beam.options.pipeline_options import PipelineOptions
+
+_HTTP_PATH = '/v1/echo'
+_PAYLOAD = base64.b64encode(bytes('payload', 'utf-8'))
+_HTTP_ENDPOINT_ADDRESS_FLAG = '--httpEndpointAddress'
+
+
+class EchoITOptions(PipelineOptions):
+ """Shared options for running integration tests on a deployed
+ ``EchoServiceGrpc`` See https://github.com/apache/beam/tree/master/.test
+ -infra/mock-apis#integration for details on how to acquire values
+ required by ``EchoITOptions``.
+ """
+ @classmethod
+ def _add_argparse_args(cls, parser) -> None:
+ parser.add_argument(
+ _HTTP_ENDPOINT_ADDRESS_FLAG,
+ dest='http_endpoint_address',
+ help='The HTTP address of the Echo API endpoint; must being with '
+ 'http(s)://')
+ parser.add_argument(
+ '--neverExceedQuotaId',
+ default='echo-should-never-exceed-quota',
+ dest='never_exceed_quota_id',
+ help='The ID for an allocated quota that should never exceed.')
+ parser.add_argument(
+ '--shouldExceedQuotaId',
+ default='echo-should-exceed-quota',
+ dest='should_exceed_quota_id',
+ help='The ID for an allocated quota that should exceed.')
+
+
+# TODO(riteshghorse,damondouglas) replace Echo(Request|Response) with proto
+# generated classes from .test-infra/mock-apis:
+@dataclass
+class EchoRequest:
+ id: str
+ payload: bytes
+
+
+@dataclass
+class EchoResponse:
+ id: str
+ payload: bytes
+
+
+class EchoHTTPCaller(Caller):
+ """Implements ``Caller`` to call the ``EchoServiceGrpc``'s HTTP handler.
+ The purpose of ``EchoHTTPCaller`` is to support integration tests.
+ """
+ def __init__(self, url: str):
+ self.url = url + _HTTP_PATH
+
+ def call(self, request: EchoRequest) -> EchoResponse:
+ """Overrides ``Caller``'s call method invoking the
+ ``EchoServiceGrpc``'s HTTP handler with an ``EchoRequest``, returning
+ either a successful ``EchoResponse`` or throwing either a
+ ``UserCodeExecutionException``, ``UserCodeTimeoutException``,
+ or a ``UserCodeQuotaException``.
+ """
+
+ try:
+ resp = urllib3.request(
+ "POST",
+ self.url,
+ json={
+ "id": request.id, "payload": str(request.payload, 'utf-8')
+ },
+ retries=False)
+
+ if resp.status < 300:
+ resp_body = resp.json()
+ resp_id = resp_body['id']
+ payload = resp_body['payload']
+ return EchoResponse(id=resp_id, payload=bytes(payload, 'utf-8'))
+
+ if resp.status == 429: # Too Many Requests
+ raise UserCodeQuotaException(resp.reason)
+
+ raise UserCodeExecutionException(resp.reason)
+
+ except urllib3.exceptions.HTTPError as e:
+ raise UserCodeExecutionException(e)
+
+
+class EchoHTTPCallerTestIT(unittest.TestCase):
+ options: Union[EchoITOptions, None] = None
+ client: Union[EchoHTTPCaller, None] = None
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.options = EchoITOptions()
+ http_endpoint_address = cls.options.http_endpoint_address
+ if not http_endpoint_address or http_endpoint_address == '':
+ raise unittest.SkipTest(f'{_HTTP_ENDPOINT_ADDRESS_FLAG} is required.')
+
+ cls.client = EchoHTTPCaller(http_endpoint_address)
+
+ def setUp(self) -> None:
Review Comment:
`setup` vs `setUp`?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "damondouglas (via GitHub)" <gi...@apache.org>.
damondouglas merged PR #29421:
URL: https://github.com/apache/beam/pull/29421
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "damondouglas (via GitHub)" <gi...@apache.org>.
damondouglas commented on PR #29421:
URL: https://github.com/apache/beam/pull/29421#issuecomment-1809039178
R: @riteshghorse
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #29421:
URL: https://github.com/apache/beam/pull/29421#issuecomment-1809271625
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: `100 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`7057ee6`)](https://app.codecov.io/gh/apache/beam/commit/7057ee634c9739c81621f130c8932684675effd7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 38.30% compared to head [(`b551eb8`)](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 38.34%.
> Report is 69 commits behind head on master.
| [Files](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [sdks/python/apache\_beam/io/requestresponseio.py](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vcmVxdWVzdHJlc3BvbnNlaW8ucHk=) | 0.00% | [19 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [sdks/python/apache\_beam/typehints/typehints.py](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHlwZWhpbnRzL3R5cGVoaW50cy5weQ==) | 34.61% | [17 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [sdks/go/pkg/beam/io/natsio/common.go](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9uYXRzaW8vY29tbW9uLmdv) | 40.90% | [10 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...o/pkg/beam/runners/prism/internal/handlecombine.go](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL3ByaXNtL2ludGVybmFsL2hhbmRsZWNvbWJpbmUuZ28=) | 46.15% | [7 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [sdks/go/pkg/beam/runners/prism/internal/stage.go](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL3ByaXNtL2ludGVybmFsL3N0YWdlLmdv) | 74.07% | [6 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [sdks/go/pkg/beam/io/natsio/write.go](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9uYXRzaW8vd3JpdGUuZ28=) | 86.95% | [4 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...s/go/pkg/beam/runners/prism/internal/preprocess.go](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL3ByaXNtL2ludGVybmFsL3ByZXByb2Nlc3MuZ28=) | 95.74% | [3 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [sdks/go/pkg/beam/io/natsio/write\_option.go](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9uYXRzaW8vd3JpdGVfb3B0aW9uLmdv) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [sdks/python/apache\_beam/transforms/external.py](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9leHRlcm5hbC5weQ==) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...go/pkg/beam/runners/prism/internal/handlerunner.go](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL3ByaXNtL2ludGVybmFsL2hhbmRsZXJ1bm5lci5nbw==) | 93.33% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [6 more](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #29421 +/- ##
==========================================
+ Coverage 38.30% 38.34% +0.03%
==========================================
Files 690 694 +4
Lines 102083 102256 +173
==========================================
+ Hits 39099 39206 +107
- Misses 61401 61460 +59
- Partials 1583 1590 +7
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/29421/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/29421/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `29.87% <19.40%> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/29421?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1396331102
##########
sdks/python/apache_beam/io/requestresponseio.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""``PTransform`` for reading from and writing to Web APIs."""
+import abc
+from typing import TypeVar
+
+RequestT = TypeVar('RequestT')
+ResponseT = TypeVar('ResponseT')
+
+
+class UserCodeExecutionException(Exception):
+ """Base class for errors related to calling Web APIs."""
+
+
+class UserCodeQuotaException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal specifically that
+ the Web API client encountered a Quota or API overuse related error.
+ """
+
+
+class UserCodeTimeoutException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal a user code timeout."""
+
+
+class Caller(metaclass=abc.ABCMeta):
+ """Interfaces user custom code intended for API calls."""
+ @abc.abstractmethod
+ def call(self, request: RequestT) -> ResponseT:
+ """Calls a Web API with the ``RequestT`` and returns a
+ ``ResponseT``. ``RequestResponseIO`` expects implementations of the
+ call method to throw either a ``UserCodeExecutionException``,
+ ``UserCodeQuotaException``, or ``UserCodeTimeoutException``.
+ """
+ pass
+
+
+class SetupTeardown(metaclass=abc.ABCMeta):
Review Comment:
looking at it..
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1391745095
##########
sdks/python/apache_beam/io/requestresponseio_it_test.py:
##########
@@ -0,0 +1,168 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import base64
+import sys
+import unittest
+from dataclasses import dataclass
+from typing import Tuple
+from typing import Union
+
+import urllib3
+
+from apache_beam.io.requestresponseio import Caller
+from apache_beam.io.requestresponseio import UserCodeExecutionException
+from apache_beam.io.requestresponseio import UserCodeQuotaException
+from apache_beam.options.pipeline_options import PipelineOptions
+
+_HTTP_PATH = "/v1/echo"
+_PAYLOAD = base64.b64encode(bytes('payload', 'utf-8'))
+
+
+class EchoITOptions(PipelineOptions):
+ """Shared options for running integration tests on a deployed
+ ``EchoServiceGrpc`` See https://github.com/apache/beam/tree/master/.test
+ -infra/mock-apis#integration for details on how to acquire values
+ required by ``EchoITOptions``.
+ """
+ @classmethod
+ def _add_argparse_args(cls, parser) -> None:
+ parser.add_argument(
+ '--httpEndpointAddress',
+ required=True,
+ dest='http_endpoint_address',
+ help='The HTTP address of the Echo API endpoint; must being with '
+ 'http(s)://')
+ parser.add_argument(
+ '--neverExceedQuotaId',
+ default='echo-should-never-exceed-quota',
+ dest='never_exceed_quota_id',
+ help='The ID for an allocated quota that should never exceed.')
+ parser.add_argument(
+ '--shouldExceedQuotaId',
+ default='echo-should-exceed-quota',
+ dest='should_exceed_quota_id',
+ help='The ID for an allocated quota that should exceed.')
+
+
+# TODO(riteshghorse,damondouglas) replace Echo(Request|Response) with proto
+# generated classes from .test-infra/mock-apis:
+@dataclass
+class EchoRequest:
+ id: str
+ payload: bytes
+
+
+@dataclass
+class EchoResponse:
+ id: str
+ payload: bytes
+
+
+class EchoHTTPCaller(Caller):
+ """Implements ``Caller`` to call the ``EchoServiceGrpc``'s HTTP handler.
+ The purpose of ``EchoHTTPCaller`` is to support integration tests.
+ """
+ def __init__(self, url: str):
+ self.url = url + _HTTP_PATH
+
+ def call(self, request: EchoRequest) -> EchoResponse:
+ """Overrides ``Caller``'s call method invoking the
+ ``EchoServiceGrpc``'s HTTP handler with an ``EchoRequest``, returning
+ either a successful ``EchoResponse`` or throwing either a
+ ``UserCodeExecutionException``, ``UserCodeTimeoutException``,
+ or a ``UserCodeQuotaException``.
+ """
+
+ try:
+ resp = urllib3.request(
+ "POST",
+ self.url,
+ json={
+ "id": request.id, "payload": str(request.payload, 'utf-8')
+ },
+ retries=False)
+
+ if resp.status < 300:
+ resp_body = resp.json()
+ resp_id = resp_body['id']
+ payload = resp_body['payload']
+ return EchoResponse(id=resp_id, payload=bytes(payload, 'utf-8'))
+
+ if resp.status == 429: # Too Many Requests
+ raise UserCodeQuotaException(resp.reason)
+
+ raise UserCodeExecutionException(resp.reason)
+
+ except urllib3.exceptions.HTTPError as e:
+ raise UserCodeExecutionException(e)
+
+
+class EchoHTTPCallerTestIT(unittest.TestCase):
+ options: Union[EchoITOptions, None] = None
+ client: Union[EchoHTTPCaller, None] = None
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ cls.options = EchoITOptions()
Review Comment:
we need to pass `httpEndpointAddress` to `EchoITOptions()` 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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1396331102
##########
sdks/python/apache_beam/io/requestresponseio.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""``PTransform`` for reading from and writing to Web APIs."""
+import abc
+from typing import TypeVar
+
+RequestT = TypeVar('RequestT')
+ResponseT = TypeVar('ResponseT')
+
+
+class UserCodeExecutionException(Exception):
+ """Base class for errors related to calling Web APIs."""
+
+
+class UserCodeQuotaException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal specifically that
+ the Web API client encountered a Quota or API overuse related error.
+ """
+
+
+class UserCodeTimeoutException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal a user code timeout."""
+
+
+class Caller(metaclass=abc.ABCMeta):
+ """Interfaces user custom code intended for API calls."""
+ @abc.abstractmethod
+ def call(self, request: RequestT) -> ResponseT:
+ """Calls a Web API with the ``RequestT`` and returns a
+ ``ResponseT``. ``RequestResponseIO`` expects implementations of the
+ call method to throw either a ``UserCodeExecutionException``,
+ ``UserCodeQuotaException``, or ``UserCodeTimeoutException``.
+ """
+ pass
+
+
+class SetupTeardown(metaclass=abc.ABCMeta):
Review Comment:
looking at it.. to be clear you mean to replace SetupTeardown with contextlib right?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [RRIO] [Call] Define Caller & SetupTeardown interfaces [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #29421:
URL: https://github.com/apache/beam/pull/29421#discussion_r1396384310
##########
sdks/python/apache_beam/io/requestresponseio.py:
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""``PTransform`` for reading from and writing to Web APIs."""
+import abc
+from typing import TypeVar
+
+RequestT = TypeVar('RequestT')
+ResponseT = TypeVar('ResponseT')
+
+
+class UserCodeExecutionException(Exception):
+ """Base class for errors related to calling Web APIs."""
+
+
+class UserCodeQuotaException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal specifically that
+ the Web API client encountered a Quota or API overuse related error.
+ """
+
+
+class UserCodeTimeoutException(UserCodeExecutionException):
+ """Extends ``UserCodeExecutionException`` to signal a user code timeout."""
+
+
+class Caller(metaclass=abc.ABCMeta):
+ """Interfaces user custom code intended for API calls."""
+ @abc.abstractmethod
+ def call(self, request: RequestT) -> ResponseT:
+ """Calls a Web API with the ``RequestT`` and returns a
+ ``ResponseT``. ``RequestResponseIO`` expects implementations of the
+ call method to throw either a ``UserCodeExecutionException``,
+ ``UserCodeQuotaException``, or ``UserCodeTimeoutException``.
+ """
+ pass
+
+
+class SetupTeardown(metaclass=abc.ABCMeta):
Review Comment:
Yeah.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org