Skip to content

fix(cdp): pass schema as runtime option and add typed join handling - #224

Merged
iht merged 2 commits into
GoogleCloudPlatform:mainfrom
adi-IL:fix/cdp-schema-loading-and-type-annotations
Aug 24, 2026
Merged

iht merged 2 commits into
GoogleCloudPlatform:mainfrom
adi-IL:fix/cdp-schema-loading-and-type-annotations

Conversation

@adi-IL

@adi-IL adi-IL commented Aug 23, 2026

Copy link
Copy Markdown
Contributor

Fixes #88, Fixes #89

What was wrong

  1. In pipelines/cdp/cdp_pipeline/customer_data_platform.py, the BigQuery output schema was read directly at module import time with a hardcoded relative path (with open("./schema/unified_table.json", encoding="utf-8")). When the package was installed on Dataflow workers or executed outside the pipelines/cdp directory, the relative path failed with FileNotFoundError.
  2. In left_join, coupon_redempt_value lacked proper type annotations and type checks, triggering linter warnings regarding dictionary subscriptability on Optional types.

What changed

  • Removed top-level schema file loading from module scope.
  • Added load_output_schema(schema_path) helper function that supports custom schema paths, package-relative fallback paths, and embedded default schema definitions.
  • Added --output_schema_path argument to MyPipelineOptions in pipelines/cdp/cdp_pipeline/options.py.
  • Updated create_and_run_pipeline to load and pass the schema dynamically at pipeline launch time.
  • Added complete type annotations and defensive dictionary checks to left_join for clean linter validation.
  • Added unit tests in pipelines/cdp/tests/test_customer_data_platform.py covering matching joins, non-matching joins, schema loading, and Beam pipeline transformation execution with TestPipeline.

Verification

  • yapf -i -r --style yapf . -> clean formatting
  • pylint --rcfile ../pylintrc cdp_pipeline tests -> 10.00/10 rating
  • pytest tests/ -> 4 passed
  • python setup.py sdist -> package builds successfully

@iht iht left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM! Verified that both #88 (schema loading at launch time with package fallback and CLI option) and #89 (type annotations, defensive dictionary handling in left_join, and unit tests) are fully resolved. All CI checks passed successfully.

@iht
iht merged commit 73f05d1 into GoogleCloudPlatform:main Aug 24, 2026
11 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

2 participants