feat(elt-pipelines): Proposal postgresql sources pipeline#396
feat(elt-pipelines): Proposal postgresql sources pipeline#396bashanlam wants to merge 7 commits into
Conversation
Implement the elt-pipeline proposal to extract data from the PostgreSQL database source and load it into the Iceberg destination.
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughChangesAdds environment-configured PostgreSQL extraction, schema mapping to PyArrow, chunked table streaming, and pipeline resources for each configured table. Postgres extraction pipeline
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
elt-pipelines/fase/ingest/fase/utils/postgres.py (2)
18-23: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFormat multiple statements on separate lines.
Putting multiple statements on a single line violates PEP 8 guidelines and reduces readability.
🧹 Proposed fix
- if 'int' in t: return 'bigint' - if 'bool' in t: return 'bool' - if 'json' in t or 'uuid' in t: return 'text' - if 'float' in t or 'numeric' in t or 'double' in t: return 'double' - if 'timestamp' in t: return 'timestamp' - if 'date' in t: return 'date' + if 'int' in t: + return 'bigint' + if 'bool' in t: + return 'bool' + if 'json' in t or 'uuid' in t: + return 'text' + if 'float' in t or 'numeric' in t or 'double' in t: + return 'double' + if 'timestamp' in t: + return 'timestamp' + if 'date' in t: + return 'date'🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@elt-pipelines/fase/ingest/fase/utils/postgres.py` around lines 18 - 23, Reformat the type-mapping conditionals in the visible utility function so each if statement and its return statement appears on separate lines, preserving the existing ordering and return values.Source: Linters/SAST tools
39-44: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFormat multiple statements on separate lines.
Placing the
if/elif/elsecondition and assignment on the same line makes the code harder to read and violates PEP 8 formatting standards.🧹 Proposed fix
- if pq_type_str == 'bigint': pa_type = pa.int64() - elif pq_type_str == 'bool': pa_type = pa.bool_() - elif pq_type_str == 'double': pa_type = pa.float64() - elif pq_type_str == 'timestamp': pa_type = pa.timestamp('us') - elif pq_type_str == 'date': pa_type = pa.date32() - else: pa_type = pa.string() + if pq_type_str == 'bigint': + pa_type = pa.int64() + elif pq_type_str == 'bool': + pa_type = pa.bool_() + elif pq_type_str == 'double': + pa_type = pa.float64() + elif pq_type_str == 'timestamp': + pa_type = pa.timestamp('us') + elif pq_type_str == 'date': + pa_type = pa.date32() + else: + pa_type = pa.string()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@elt-pipelines/fase/ingest/fase/utils/postgres.py` around lines 39 - 44, Reformat the type-mapping conditional in the visible PostgreSQL utility code so each if/elif/else condition and its pa_type assignment appear on separate properly indented lines, without changing the existing type mappings or fallback behavior.Source: Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@elt-pipelines/fase/ingest/fase/proposal/proposal.py`:
- Around line 31-37: Update the connection_uri method to URL-encode the stripped
username and password with urllib.parse.quote_plus before interpolating them
into the URI, while preserving the existing driver, host, port, and database
formatting.
In `@elt-pipelines/fase/ingest/fase/utils/postgres.py`:
- Around line 81-84: Update the empty-result branch around has_data so it yields
a zero-row PyArrow table using target_schema, rather than constructing dummy_row
and a one-row DataFrame of None values. Preserve the existing behavior for
sources containing data.
---
Nitpick comments:
In `@elt-pipelines/fase/ingest/fase/utils/postgres.py`:
- Around line 18-23: Reformat the type-mapping conditionals in the visible
utility function so each if statement and its return statement appears on
separate lines, preserving the existing ordering and return values.
- Around line 39-44: Reformat the type-mapping conditional in the visible
PostgreSQL utility code so each if/elif/else condition and its pa_type
assignment appear on separate properly indented lines, without changing the
existing type mappings or fallback behavior.
🪄 Autofix (Beta)
✅ Autofix completed
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: bf9112f6-fe76-4ea3-a46b-f6dde01e1c47
📒 Files selected for processing (2)
elt-pipelines/fase/ingest/fase/proposal/proposal.pyelt-pipelines/fase/ingest/fase/utils/postgres.py
bashanlam
changed the title
bashanlam
changed the title
|
Note Autofix is a beta feature. Expect some limitations and changes as we gather feedback and continue to improve it. Fixes Applied SuccessfullyFixed 2 file(s) based on 2 unresolved review comments. Files modified:
Commit: The changes have been pushed to the Time taken: |
Fixed 2 file(s) based on 2 unresolved review comments. Co-authored-by: CodeRabbit <noreply@coderabbit.ai>
This reverts commit 62abf0d.
WHTaylor
left a comment
WHTaylor
left a comment
There was a problem hiding this comment.
A lot of this was supposed to be covered by the sqldatabase package. If that didn't work for this, it'd probably be ideal if we could fix it instead of reimplementing a lot of the same functionality.
Add set up for a fase_landing warehouse in the local infrastructure Centralize loading pg and oracle by using sqlalchemy at sqldatabase/__init__.py
3 participants
ref #340
Description
This PR implements the ingestion elt-pipelines to move proposal application data into the lakehouse. It establishes the end-to-end data flow by extracting data from PostgreSQL sources and loading it into the Iceberg destination.
Scope of Work
Summary by CodeRabbit