You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "mohamedawnallah (via GitHub)" <gi...@apache.org> on 2025/06/29 02:44:53 UTC
[PR] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
mohamedawnallah opened a new pull request, #35473:
URL: https://github.com/apache/beam/pull/35473
## Description
- [x] Update the [Enrichment Documentation](https://beam.apache.org/documentation/transforms/python/elementwise/enrichment/) to include Google CloudSQL Enrichment Handler integration
- [x] Add Python Example for Google CloudSQL (Managed PostgreSQL) Enrichment to the Documentation
- [x] Add Python Example for Unmanaged PostgreSQL Enrichment to the Documentation
- [x] Add Python Example for Unmanaged MySQL Enrichment to the Documentation
- [x] Add Python Example for Unmanaged Microsoft SQL Server Enrichment to the Documentation
Closes #30773.
------------------------
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3016283034
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2332204148
##########
sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:
##########
@@ -60,52 +82,227 @@ def validate_enrichment_with_vertex_ai_legacy():
return expected
-def std_out_to_dict(stdout_lines, row_key):
- output_dict = {}
- for stdout_line in stdout_lines:
- # parse the stdout in a dictionary format so that it can be
- # evaluated/compared as one. This allows us to compare without
- # considering the order of the stdout or the order that the fields of the
- # row are arranged in.
- fmtd = '{\"' + stdout_line[4:-1].replace('=', '\": ').replace(
- ', ', ', \"').replace('\"\'', '\'') + "}"
- stdout_dict = eval(fmtd) # pylint: disable=eval-used
- output_dict[stdout_dict[row_key]] = stdout_dict
- return output_dict
+def validate_enrichment_with_google_cloudsql_pg():
+ expected = '''[START enrichment_with_google_cloudsql_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_google_cloudsql_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_pg():
+ expected = '''[START enrichment_with_external_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_mysql():
+ expected = '''[START enrichment_with_external_mysql]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_mysql]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_sqlserver():
+ expected = '''[START enrichment_with_external_sqlserver]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_sqlserver]'''.splitlines()[1:-1]
+ return expected
@mock.patch('sys.stdout', new_callable=StringIO)
+@pytest.mark.uses_testcontainer
class EnrichmentTest(unittest.TestCase):
def test_enrichment_with_bigtable(self, mock_stdout):
enrichment_with_bigtable()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_bigtable()
-
- self.assertEqual(len(output), len(expected))
- self.assertEqual(
- std_out_to_dict(output, 'sale_id'),
- std_out_to_dict(expected, 'sale_id'))
+ self.assertEqual(output, expected)
def test_enrichment_with_vertex_ai(self, mock_stdout):
enrichment_with_vertex_ai()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_vertex_ai()
- self.assertEqual(len(output), len(expected))
- self.assertEqual(
- std_out_to_dict(output, 'user_id'),
- std_out_to_dict(expected, 'user_id'))
+ for i in range(len(expected)):
+ self.assertEqual(set(output[i].split(',')), set(expected[i].split(',')))
def test_enrichment_with_vertex_ai_legacy(self, mock_stdout):
enrichment_with_vertex_ai_legacy()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_vertex_ai_legacy()
self.maxDiff = None
+ self.assertEqual(output, expected)
+
+ def test_enrichment_with_google_cloudsql_pg(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.POSTGRESQL
+ with EnrichmentTestHelpers.sql_test_context(True, db_adapter):
+ try:
+ enrichment_with_google_cloudsql_pg()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_google_cloudsql_pg()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+ def test_enrichment_with_external_pg(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.POSTGRESQL
+ with EnrichmentTestHelpers.sql_test_context(False, db_adapter):
+ try:
+ enrichment_with_external_pg()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_external_pg()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+ def test_enrichment_with_external_mysql(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.MYSQL
+ with EnrichmentTestHelpers.sql_test_context(False, db_adapter):
+ try:
+ enrichment_with_external_mysql()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_external_mysql()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+ def test_enrichment_with_external_sqlserver(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.SQLSERVER
+ with EnrichmentTestHelpers.sql_test_context(False, db_adapter):
+ try:
+ enrichment_with_external_sqlserver()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_external_sqlserver()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+
+@dataclass
+class CloudSQLEnrichmentTestDataConstruct:
+ client_handler: Callable[[], DBAPIConnection]
+ engine: Engine
+ metadata: MetaData
+ db: SQLDBContainerInfo = None
+
+
+class EnrichmentTestHelpers:
+ @contextmanager
+ def sql_test_context(is_cloudsql: bool, db_adapter: DatabaseTypeAdapter):
+ result: Optional[CloudSQLEnrichmentTestDataConstruct] = None
+ try:
+ result = EnrichmentTestHelpers.pre_sql_enrichment_test(
+ is_cloudsql, db_adapter)
+ yield
+ finally:
+ if result:
+ EnrichmentTestHelpers.post_sql_enrichment_test(result)
+
+ @staticmethod
+ def pre_sql_enrichment_test(
+ is_cloudsql: bool,
+ db_adapter: DatabaseTypeAdapter) -> CloudSQLEnrichmentTestDataConstruct:
+ table_id = "products"
+ columns = [
+ Column("product_id", Integer, primary_key=True),
+ Column("name", VARCHAR(255), nullable=False),
+ Column("quantity", Integer, nullable=False),
+ Column("region_id", Integer, nullable=False),
+ ]
+ table_data = [
+ {
+ "product_id": 1, "name": "A", 'quantity': 2, 'region_id': 3
+ },
+ {
+ "product_id": 2, "name": "B", 'quantity': 3, 'region_id': 1
+ },
+ {
+ "product_id": 3, "name": "C", 'quantity': 10, 'region_id': 4
+ },
+ ]
+ metadata = MetaData()
+
+ connection_config: ConnectionConfig
+ if is_cloudsql:
+ gcp_project_id = "apache-beam-testing"
+ region = "us-central1"
+ instance_name = "beam-integration-tests"
+ instance_connection_uri = f"{gcp_project_id}:{region}:{instance_name}"
+ db_id = "postgres"
+ user = "postgres"
+ password = os.getenv("ALLOYDB_PASSWORD")
+ os.environ['GOOGLE_CLOUD_SQL_DB_URI'] = instance_connection_uri
+ os.environ['GOOGLE_CLOUD_SQL_DB_ID'] = db_id
+ os.environ['GOOGLE_CLOUD_SQL_DB_USER'] = user
+ os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
+ os.environ['GOOGLE_CLOUD_SQL_DB_TABLE_ID'] = table_id
+ connection_config = CloudSQLConnectionConfig(
+ db_adapter=db_adapter,
+ instance_connection_uri=instance_connection_uri,
+ user=user,
+ password=password,
+ db_id=db_id)
+ else:
+ db = SQLEnrichmentTestHelper.start_sql_db_container(db_adapter)
Review Comment:
> I think we need something like this in both paths. The new error I'm seeing in CI is:
>
> ```
> SQLEnrichmentTestHelper.create_table(
> table_id=table_id,
> engine=engine,
> columns=columns,
> table_data=table_data,
> metadata=metadata)
>
> result = CloudSQLEnrichmentTestDataConstruct(
> > db=db, client_handler=conenctor, engine=engine, metadata=metadata)
> E UnboundLocalError: local variable 'db' referenced before assignment
> ```
>
> Looks like the pass-through config did work though
Fixed that unbounded local variable
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3270064808
We can enforce running CloudSQL enrichment handler on Precommit Python Transforms by removing `is_integration_test=True` from `TestPipeline` and removing using `uses_testcontainer` pytest marker so it not collected to run in the postcommit suite. That way we can limit the scope of running those test cases as well as avoiding such flaky runs on postcommit suite https://github.com/apache/beam/pull/35816#issuecomment-3269388436:
https://github.com/apache/beam/blob/a2f9fb8e99241313bf3a0957a6aac9c24a8217b3/sdks/python/test-suites/direct/common.gradle#L374-L393
That require us to add `ALLOYDB_PASSWORD` env variable to `beam_PreCommit_Python_Transforms` workflow so Google CloudSQL PostgreSQL instance can run there without being skipped
https://github.com/apache/beam/blob/a2f9fb8e99241313bf3a0957a6aac9c24a8217b3/.github/workflows/beam_PreCommit_Python_Transforms.yml#L52-L56
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3258257708
Looks like the coverage failure here is probably a real issue. We probably need the same change there that we're making to the examples workflow.
```
<testcase classname="apache_beam.examples.snippets.transforms.elementwise.enrichment_test.EnrichmentTest" name="test_enrichment_with_google_cloudsql_pg" time="2.420">
<failure message="TypeError: str expected, not NoneType">self = <apache_beam.examples.snippets.transforms.elementwise.enrichment_test.EnrichmentTest testMethod=test_enrichment_with_google_cloudsql_pg> mock_stdout = <_io.StringIO object at 0x781158679f70> def test_enrichment_with_google_cloudsql_pg(self, mock_stdout): db_adapter = DatabaseTypeAdapter.POSTGRESQL > with EnrichmentTestHelpers.sql_test_context(True, db_adapter): apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:147: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ /opt/hostedtoolcache/Python/3.9.22/x64/lib/python3.9/contextlib.py:119: in __enter__ return next(self.gen) apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:203: in sql_test_context result = EnrichmentTestHelpers.pre_sql_enrichment_test( apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:246: in pre_sql_enrichment_test os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = p
assword /opt/hostedtoolcache/Python/3.9.22/x64/lib/python3.9/os.py:684: in __setitem__ value = self.encodevalue(value) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ value = None def encode(value): if not isinstance(value, str): > raise TypeError("str expected, not %s" % type(value).__name__) E TypeError: str expected, not NoneType /opt/hostedtoolcache/Python/3.9.22/x64/lib/python3.9/os.py:756: TypeError</failure>
</testcase>
```
Rather than bundling those changes with this PR, lets make them in a separate PR so that we can confirm all tests pass here. Added https://github.com/apache/beam/pull/36061 to do this
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3272000870
> Hm, looks like the infra is right now, but the tests are failing - https://github.com/apache/beam/actions/runs/17588014826/attempts/1?pr=35473 is an example. I tried kicking it off again, but it looks like a real problem with duplicate data
Yeah looked at it debugging it now 👍
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3270112866
Submitted this PR https://github.com/apache/beam/pull/36096 to add `ALLOYDB_PASSWORD` to Precommit Python Transforms workflow
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3271218574
Merged now the changes in this PR after the merge of https://github.com/apache/beam/pull/36096
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3263935880
> @mohamedawnallah would you mind taking a look at the failures here?
>
> ```
> apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:203: in sql_test_context
> result = EnrichmentTestHelpers.pre_sql_enrichment_test(
> apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:246: in pre_sql_enrichment_test
> os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
> /opt/hostedtoolcache/Python/3.9.22/x64/lib/python3.9/os.py:684: in __setitem__
> value = self.encodevalue(value)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
>
> value = None
>
> def encode(value):
> if not isinstance(value, str):
> > raise TypeError("str expected, not %s" % type(value).__name__)
> E TypeError: str expected, not NoneType
> ```
@damccorm – The tests are currently failing because `ALLOYDB_PASSWORD` is being perceived as `None`, even though it's configured as a secret in the relevant workflows `PreCommit Python Coverage` and `PreCommit Python Examples` (https://github.com/apache/beam/pull/36061).
I've encountered this exact issue before and couldn't really figure it out (similar problem when tried to access on beam PostCommit Python workflow in https://github.com/apache/beam/pull/34398#discussion_r2279981586). This secret specifically seems having issues when attempting to use in other workflows.
The failure occurs at this specific line where the code attempts to read the password environment variable:
```python
password = os.getenv("ALLOYDB_PASSWORD")
```
https://github.com/mohamedawnallah/beam/blob/af8e4fe198b89a0abfd9a88fb2c1ab0303338772/sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py#L242
When the environment variable doesn't exist, `os.getenv()` returns `None`, causing the test to fail.
This appears to be an infrastructure issue related to how the `ALLOYDB_PASSWORD` secret is accessed across different
workflows. Perhaps @Amar3tto, @A1K28 can help
What do you think, @damccorm?
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3271223872
It looks like the coverage workflow was failing with the same issue as https://github.com/apache/beam/pull/35473#issuecomment-3259375214
Since this test is running in some suites, could we just skip it when `GOOGLE_CLOUD_SQL_DB_PASSWORD` is not present?
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2308207561
##########
.github/workflows/beam_PreCommit_Python_Examples.yml:
##########
@@ -53,6 +53,7 @@ env:
DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }}
GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }}
GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
+ ALLOYDB_PASSWORD: ${{ secrets.ALLOYDB_PASSWORD }}
Review Comment:
The CloudSQL instance on Google Cloud expected to fail in the examples since that workflow change haven't yet switched on. As seeing in the CI logs:
```
____________ EnrichmentTest.test_enrichment_with_google_cloudsql_pg ____________
[gw0] linux -- Python 3.11.12 /runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/target/.tox-py311-cloud/py311-cloud/bin/python
self = <apache_beam.examples.snippets.transforms.elementwise.enrichment_test.EnrichmentTest testMethod=test_enrichment_with_google_cloudsql_pg>
mock_stdout = <_io.StringIO object at 0x7f388c1adbd0>
def test_enrichment_with_google_cloudsql_pg(self, mock_stdout):
db_adapter = DatabaseTypeAdapter.POSTGRESQL
> with EnrichmentTestHelpers.sql_test_context(True, db_adapter):
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:147:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/opt/hostedtoolcache/Python/3.11.12/x64/lib/python3.11/contextlib.py:137: in __enter__
return next(self.gen)
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:203: in sql_test_context
result = EnrichmentTestHelpers.pre_sql_enrichment_test(
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:246: in pre_sql_enrichment_test
os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
<frozen os>:684: in __setitem__
???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = None
> ???
E TypeError: str expected, not NoneType
<frozen os>:758: TypeError
- generated xml file: /runner/_work/beam/beam/sdks/python/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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3270067442
We can enforce running CloudSQL enrichment handler on Precommit Python Transforms by removing `is_integration_test=True` from `TestPipeline` and removing using `uses_testcontainer` pytest marker so it not collected to run in the postcommit suite. That way we can limit the scope of running those test cases as well as avoiding such flaky runs on postcommit suite https://github.com/apache/beam/pull/35816#issuecomment-3269388436:
https://github.com/apache/beam/blob/a2f9fb8e99241313bf3a0957a6aac9c24a8217b3/sdks/python/test-suites/direct/common.gradle#L374-L393
That require us to add `ALLOYDB_PASSWORD` env variable to `beam_PreCommit_Python_Transforms` workflow so Google CloudSQL PostgreSQL instance can run there without being skipped
https://github.com/apache/beam/blob/a2f9fb8e99241313bf3a0957a6aac9c24a8217b3/.github/workflows/beam_PreCommit_Python_Transforms.yml#L52-L56
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3271996740
Hm, looks like the infra is right now, but the tests are failing - https://github.com/apache/beam/actions/runs/17588014826/attempts/1?pr=35473 is an example. I tried kicking it off again, but it looks like a real problem with duplicate data
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3259375214
@mohamedawnallah would you mind taking a look at the failures here?
```
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:203: in sql_test_context
result = EnrichmentTestHelpers.pre_sql_enrichment_test(
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:246: in pre_sql_enrichment_test
os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
/opt/hostedtoolcache/Python/3.9.22/x64/lib/python3.9/os.py:684: in __setitem__
value = self.encodevalue(value)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = None
def encode(value):
if not isinstance(value, str):
> raise TypeError("str expected, not %s" % type(value).__name__)
E TypeError: str expected, not NoneType
```
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm merged PR #35473:
URL: https://github.com/apache/beam/pull/35473
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3271274811
> It looks like the coverage workflow was failing with the same issue as [#35473 (comment)](https://github.com/apache/beam/pull/35473#issuecomment-3259375214)
>
> Since this test is running in some suites, could we just skip it when `GOOGLE_CLOUD_SQL_DB_PASSWORD` is not present?
Added that conditional skip 👍
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3270211932
After enforcing CloudSQL tests to run only on precommit python transforms they are running and passing except the Google CloudSQL Enrichment tests due to lack of `ALLOYDB_PASSWORD` in PreCommit Python Transforms. With this PR #36096 we can enable them to run on this workflow
```
[gw3] [ 91%] SKIPPED apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment
apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_batched
[gw3] [ 91%] SKIPPED apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_batched
apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_batched_multiple_fields
[gw3] [ 91%] SKIPPED apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_batched_multiple_fields
apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_on_non_existent_table
[gw3] [ 91%] SKIPPED apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_on_non_existent_table
apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_with_condition_value_fn
[gw3] [ 91%] SKIPPED apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_with_condition_value_fn
apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py::TestCloudSQLPostgresEnrichment::test_sql_enrichment_with_query_fn
[gw3] [ 91%] SKIPPED
...
```
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3266474895
> > @mohamedawnallah would you mind taking a look at the failures here?
> > ```
> > apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:203: in sql_test_context
> > result = EnrichmentTestHelpers.pre_sql_enrichment_test(
> > apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:246: in pre_sql_enrichment_test
> > os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
> > /opt/hostedtoolcache/Python/3.9.22/x64/lib/python3.9/os.py:684: in __setitem__
> > value = self.encodevalue(value)
> > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> >
> > value = None
> >
> > def encode(value):
> > if not isinstance(value, str):
> > > raise TypeError("str expected, not %s" % type(value).__name__)
> > E TypeError: str expected, not NoneType
> > ```
>
> The tests are currently failing because `ALLOYDB_PASSWORD` is being perceived as `None`, even though it's configured as a secret in the relevant workflows `PreCommit Python Coverage` and `PreCommit Python Examples` (#36061).
>
> I've encountered this exact issue before and couldn't really figure it out (similar problem when tried to access on beam PostCommit Python workflow in [#34398 (comment)](https://github.com/apache/beam/pull/34398#discussion_r2279981586)). This secret specifically seems having issues when attempting to use in other workflows.
>
> The failure occurs at this specific line where the code attempts to read the password environment variable:
>
> ```python
> password = os.getenv("ALLOYDB_PASSWORD")
> ```
>
> https://github.com/mohamedawnallah/beam/blob/af8e4fe198b89a0abfd9a88fb2c1ab0303338772/sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py#L242
>
> When the environment variable doesn't exist, `os.getenv()` returns `None`, causing the test to fail.
>
> This appears to be an infrastructure issue related to how the `ALLOYDB_PASSWORD` secret is accessed across different workflows. Perhaps @Amar3tto, @A1K28 can help
>
> Probably the least resistant path would be to skip these tests when `ALLOYDB_PASSWORD` is not accessible. However, this approach doesn't address the root cause, perhaps not a clean solution, and at the same time allow us to make progress on this PR and that secret access issue can be resolved later
>
> What do you think, @damccorm?
Have you tried running the gradle task that is failing locally? One idea is that it may be that we're invoking these tests with tox and need to allow list this variable to be passed through to the tox environment - https://github.com/apache/beam/blob/32f5be6c36c17ae87e1b3dc11426b95c352d3584/sdks/python/tox.ini#L34
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2308207561
##########
.github/workflows/beam_PreCommit_Python_Examples.yml:
##########
@@ -53,6 +53,7 @@ env:
DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }}
GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }}
GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
+ ALLOYDB_PASSWORD: ${{ secrets.ALLOYDB_PASSWORD }}
Review Comment:
The Google CloudSQL instance expected to fail in the examples since that workflow change haven't yet switched on. As seeing in the CI logs:
```
____________ EnrichmentTest.test_enrichment_with_google_cloudsql_pg ____________
[gw0] linux -- Python 3.11.12 /runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/target/.tox-py311-cloud/py311-cloud/bin/python
self = <apache_beam.examples.snippets.transforms.elementwise.enrichment_test.EnrichmentTest testMethod=test_enrichment_with_google_cloudsql_pg>
mock_stdout = <_io.StringIO object at 0x7f388c1adbd0>
def test_enrichment_with_google_cloudsql_pg(self, mock_stdout):
db_adapter = DatabaseTypeAdapter.POSTGRESQL
> with EnrichmentTestHelpers.sql_test_context(True, db_adapter):
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:147:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/opt/hostedtoolcache/Python/3.11.12/x64/lib/python3.11/contextlib.py:137: in __enter__
return next(self.gen)
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:203: in sql_test_context
result = EnrichmentTestHelpers.pre_sql_enrichment_test(
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:246: in pre_sql_enrichment_test
os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
<frozen os>:684: in __setitem__
???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = None
> ???
E TypeError: str expected, not NoneType
<frozen os>:758: TypeError
- generated xml file: /runner/_work/beam/beam/sdks/python/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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2308207561
##########
.github/workflows/beam_PreCommit_Python_Examples.yml:
##########
@@ -53,6 +53,7 @@ env:
DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }}
GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }}
GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
+ ALLOYDB_PASSWORD: ${{ secrets.ALLOYDB_PASSWORD }}
Review Comment:
The CloudSQL expected to fail in the examples since that workflow change haven't yet switched on. As seeing in the CI logs:
```
____________ EnrichmentTest.test_enrichment_with_google_cloudsql_pg ____________
[gw0] linux -- Python 3.11.12 /runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/target/.tox-py311-cloud/py311-cloud/bin/python
self = <apache_beam.examples.snippets.transforms.elementwise.enrichment_test.EnrichmentTest testMethod=test_enrichment_with_google_cloudsql_pg>
mock_stdout = <_io.StringIO object at 0x7f388c1adbd0>
def test_enrichment_with_google_cloudsql_pg(self, mock_stdout):
db_adapter = DatabaseTypeAdapter.POSTGRESQL
> with EnrichmentTestHelpers.sql_test_context(True, db_adapter):
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:147:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/opt/hostedtoolcache/Python/3.11.12/x64/lib/python3.11/contextlib.py:137: in __enter__
return next(self.gen)
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:203: in sql_test_context
result = EnrichmentTestHelpers.pre_sql_enrichment_test(
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:246: in pre_sql_enrichment_test
os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
<frozen os>:684: in __setitem__
???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = None
> ???
E TypeError: str expected, not NoneType
<frozen os>:758: TypeError
- generated xml file: /runner/_work/beam/beam/sdks/python/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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3233898952
@mohamedawnallah would you mind fixing the conflicts here? Otherwise I think this should be good to merge
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2279639903
##########
website/www/site/content/en/documentation/transforms/python/elementwise/enrichment-cloudsql.md:
##########
@@ -0,0 +1,146 @@
+---
+title: "Enrichment with CloudSQL"
+---
+<!--
+Licensed 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.
+-->
+
+# Use CloudSQL to enrich data
+
+{{< localstorage language language-py >}}
+
+<table>
+ <tr>
+ <td>
+ <a>
+ {{< button-pydoc path="apache_beam.transforms.enrichment_handlers.cloudsql" class="CloudSQLEnrichmentHandler" >}}
+ </a>
+ </td>
+ </tr>
+</table>
+
+Starting with Apache Beam 2.67.0, the enrichment transform includes
Review Comment:
```suggestion
Starting with Apache Beam 2.68.0, the enrichment transform includes
```
Nit since we've moved to the next release cut.
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2331356785
##########
sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:
##########
@@ -60,52 +82,227 @@ def validate_enrichment_with_vertex_ai_legacy():
return expected
-def std_out_to_dict(stdout_lines, row_key):
- output_dict = {}
- for stdout_line in stdout_lines:
- # parse the stdout in a dictionary format so that it can be
- # evaluated/compared as one. This allows us to compare without
- # considering the order of the stdout or the order that the fields of the
- # row are arranged in.
- fmtd = '{\"' + stdout_line[4:-1].replace('=', '\": ').replace(
- ', ', ', \"').replace('\"\'', '\'') + "}"
- stdout_dict = eval(fmtd) # pylint: disable=eval-used
- output_dict[stdout_dict[row_key]] = stdout_dict
- return output_dict
+def validate_enrichment_with_google_cloudsql_pg():
+ expected = '''[START enrichment_with_google_cloudsql_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_google_cloudsql_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_pg():
+ expected = '''[START enrichment_with_external_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_mysql():
+ expected = '''[START enrichment_with_external_mysql]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_mysql]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_sqlserver():
+ expected = '''[START enrichment_with_external_sqlserver]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_sqlserver]'''.splitlines()[1:-1]
+ return expected
@mock.patch('sys.stdout', new_callable=StringIO)
+@pytest.mark.uses_testcontainer
class EnrichmentTest(unittest.TestCase):
def test_enrichment_with_bigtable(self, mock_stdout):
enrichment_with_bigtable()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_bigtable()
-
- self.assertEqual(len(output), len(expected))
- self.assertEqual(
- std_out_to_dict(output, 'sale_id'),
- std_out_to_dict(expected, 'sale_id'))
+ self.assertEqual(output, expected)
def test_enrichment_with_vertex_ai(self, mock_stdout):
enrichment_with_vertex_ai()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_vertex_ai()
- self.assertEqual(len(output), len(expected))
- self.assertEqual(
- std_out_to_dict(output, 'user_id'),
- std_out_to_dict(expected, 'user_id'))
+ for i in range(len(expected)):
+ self.assertEqual(set(output[i].split(',')), set(expected[i].split(',')))
def test_enrichment_with_vertex_ai_legacy(self, mock_stdout):
enrichment_with_vertex_ai_legacy()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_vertex_ai_legacy()
self.maxDiff = None
+ self.assertEqual(output, expected)
+
+ def test_enrichment_with_google_cloudsql_pg(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.POSTGRESQL
+ with EnrichmentTestHelpers.sql_test_context(True, db_adapter):
+ try:
+ enrichment_with_google_cloudsql_pg()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_google_cloudsql_pg()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+ def test_enrichment_with_external_pg(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.POSTGRESQL
+ with EnrichmentTestHelpers.sql_test_context(False, db_adapter):
+ try:
+ enrichment_with_external_pg()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_external_pg()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+ def test_enrichment_with_external_mysql(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.MYSQL
+ with EnrichmentTestHelpers.sql_test_context(False, db_adapter):
+ try:
+ enrichment_with_external_mysql()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_external_mysql()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+ def test_enrichment_with_external_sqlserver(self, mock_stdout):
+ db_adapter = DatabaseTypeAdapter.SQLSERVER
+ with EnrichmentTestHelpers.sql_test_context(False, db_adapter):
+ try:
+ enrichment_with_external_sqlserver()
+ output = mock_stdout.getvalue().splitlines()
+ expected = validate_enrichment_with_external_sqlserver()
+ self.assertEqual(output, expected)
+ except Exception as e:
+ self.fail(f"Test failed with unexpected error: {e}")
+
+
+@dataclass
+class CloudSQLEnrichmentTestDataConstruct:
+ client_handler: Callable[[], DBAPIConnection]
+ engine: Engine
+ metadata: MetaData
+ db: SQLDBContainerInfo = None
+
+
+class EnrichmentTestHelpers:
+ @contextmanager
+ def sql_test_context(is_cloudsql: bool, db_adapter: DatabaseTypeAdapter):
+ result: Optional[CloudSQLEnrichmentTestDataConstruct] = None
+ try:
+ result = EnrichmentTestHelpers.pre_sql_enrichment_test(
+ is_cloudsql, db_adapter)
+ yield
+ finally:
+ if result:
+ EnrichmentTestHelpers.post_sql_enrichment_test(result)
+
+ @staticmethod
+ def pre_sql_enrichment_test(
+ is_cloudsql: bool,
+ db_adapter: DatabaseTypeAdapter) -> CloudSQLEnrichmentTestDataConstruct:
+ table_id = "products"
+ columns = [
+ Column("product_id", Integer, primary_key=True),
+ Column("name", VARCHAR(255), nullable=False),
+ Column("quantity", Integer, nullable=False),
+ Column("region_id", Integer, nullable=False),
+ ]
+ table_data = [
+ {
+ "product_id": 1, "name": "A", 'quantity': 2, 'region_id': 3
+ },
+ {
+ "product_id": 2, "name": "B", 'quantity': 3, 'region_id': 1
+ },
+ {
+ "product_id": 3, "name": "C", 'quantity': 10, 'region_id': 4
+ },
+ ]
+ metadata = MetaData()
+
+ connection_config: ConnectionConfig
+ if is_cloudsql:
+ gcp_project_id = "apache-beam-testing"
+ region = "us-central1"
+ instance_name = "beam-integration-tests"
+ instance_connection_uri = f"{gcp_project_id}:{region}:{instance_name}"
+ db_id = "postgres"
+ user = "postgres"
+ password = os.getenv("ALLOYDB_PASSWORD")
+ os.environ['GOOGLE_CLOUD_SQL_DB_URI'] = instance_connection_uri
+ os.environ['GOOGLE_CLOUD_SQL_DB_ID'] = db_id
+ os.environ['GOOGLE_CLOUD_SQL_DB_USER'] = user
+ os.environ['GOOGLE_CLOUD_SQL_DB_PASSWORD'] = password
+ os.environ['GOOGLE_CLOUD_SQL_DB_TABLE_ID'] = table_id
+ connection_config = CloudSQLConnectionConfig(
+ db_adapter=db_adapter,
+ instance_connection_uri=instance_connection_uri,
+ user=user,
+ password=password,
+ db_id=db_id)
+ else:
+ db = SQLEnrichmentTestHelper.start_sql_db_container(db_adapter)
Review Comment:
I think we need something like this in both paths. The new error I'm seeing in CI is:
```
SQLEnrichmentTestHelper.create_table(
table_id=table_id,
engine=engine,
columns=columns,
table_data=table_data,
metadata=metadata)
result = CloudSQLEnrichmentTestDataConstruct(
> db=db, client_handler=conenctor, engine=engine, metadata=metadata)
E UnboundLocalError: local variable 'db' referenced before assignment
```
Looks like the pass-through config did work though
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on PR #35473:
URL: https://github.com/apache/beam/pull/35473#issuecomment-3273282666
@damccorm - the relevant CI tests are green now
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2308217181
##########
sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:
##########
@@ -60,52 +82,227 @@ def validate_enrichment_with_vertex_ai_legacy():
return expected
-def std_out_to_dict(stdout_lines, row_key):
- output_dict = {}
- for stdout_line in stdout_lines:
- # parse the stdout in a dictionary format so that it can be
- # evaluated/compared as one. This allows us to compare without
- # considering the order of the stdout or the order that the fields of the
- # row are arranged in.
- fmtd = '{\"' + stdout_line[4:-1].replace('=', '\": ').replace(
- ', ', ', \"').replace('\"\'', '\'') + "}"
- stdout_dict = eval(fmtd) # pylint: disable=eval-used
- output_dict[stdout_dict[row_key]] = stdout_dict
- return output_dict
+def validate_enrichment_with_google_cloudsql_pg():
+ expected = '''[START enrichment_with_google_cloudsql_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_google_cloudsql_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_pg():
+ expected = '''[START enrichment_with_external_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_mysql():
+ expected = '''[START enrichment_with_external_mysql]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_mysql]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_sqlserver():
+ expected = '''[START enrichment_with_external_sqlserver]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_sqlserver]'''.splitlines()[1:-1]
+ return expected
@mock.patch('sys.stdout', new_callable=StringIO)
+@pytest.mark.uses_testcontainer
Review Comment:
It is nice that those Milvus testcontainers runs on that workflow environment without requiring `docker_in_docker`. Not sure exactly what is the variable parameter (other than different workflow) that makes them really pass here given they are underlying util for creating/stopping docker container!
--
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] [2/2] sdks/python: enrich data with CloudSQL [PostgreSQL, MySQL, SQLServer] [beam]
Posted by "mohamedawnallah (via GitHub)" <gi...@apache.org>.
mohamedawnallah commented on code in PR #35473:
URL: https://github.com/apache/beam/pull/35473#discussion_r2308217181
##########
sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:
##########
@@ -60,52 +82,227 @@ def validate_enrichment_with_vertex_ai_legacy():
return expected
-def std_out_to_dict(stdout_lines, row_key):
- output_dict = {}
- for stdout_line in stdout_lines:
- # parse the stdout in a dictionary format so that it can be
- # evaluated/compared as one. This allows us to compare without
- # considering the order of the stdout or the order that the fields of the
- # row are arranged in.
- fmtd = '{\"' + stdout_line[4:-1].replace('=', '\": ').replace(
- ', ', ', \"').replace('\"\'', '\'') + "}"
- stdout_dict = eval(fmtd) # pylint: disable=eval-used
- output_dict[stdout_dict[row_key]] = stdout_dict
- return output_dict
+def validate_enrichment_with_google_cloudsql_pg():
+ expected = '''[START enrichment_with_google_cloudsql_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_google_cloudsql_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_pg():
+ expected = '''[START enrichment_with_external_pg]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_pg]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_mysql():
+ expected = '''[START enrichment_with_external_mysql]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_mysql]'''.splitlines()[1:-1]
+ return expected
+
+
+def validate_enrichment_with_external_sqlserver():
+ expected = '''[START enrichment_with_external_sqlserver]
+Row(product_id=1, name='A', quantity=2, region_id=3)
+Row(product_id=2, name='B', quantity=3, region_id=1)
+Row(product_id=3, name='C', quantity=10, region_id=4)
+ [END enrichment_with_external_sqlserver]'''.splitlines()[1:-1]
+ return expected
@mock.patch('sys.stdout', new_callable=StringIO)
+@pytest.mark.uses_testcontainer
Review Comment:
It is nice that those Milvus testcontainers runs on that workflow environment without requiring `docker_in_docker`. Not sure exactly what is the variable parameter (other than different workflow) that makes them really pass here given they are underlying util for creating/stopping docker container!
--
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