feat(elt-pipelines): Proposal postgresql sources pipeline#396
feat(elt-pipelines): Proposal postgresql sources pipeline#396bashanlam wants to merge 4 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 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.
| class Extract(BaseExtract): | ||
| def __init__(self, *args, **kwargs): | ||
| super().__init__(*args, **kwargs) | ||
| # Instantiate pipeline config and pass it directly to the generic utility extractor | ||
| self.postgres_config = PipelinePostgresConfig() | ||
| self.extractor = PostgresExtractor(self.postgres_config) |
There was a problem hiding this comment.
Extract classes are instantiated in such a way that doing this:
| class Extract(BaseExtract): | |
| def __init__(self, *args, **kwargs): | |
| super().__init__(*args, **kwargs) | |
| # Instantiate pipeline config and pass it directly to the generic utility extractor | |
| self.postgres_config = PipelinePostgresConfig() | |
| self.extractor = PostgresExtractor(self.postgres_config) | |
| class Extract(BaseExtract): | |
| config_cls = PipelinePostgresConfig | |
| def __init__(self, config: PipelinePostgresConfig): | |
| super().__init__(config) | |
| self.extractor = PostgresExtractor(config) |
Will set the config up correctly, and make it so the environment variables are prefixed with the job name (so they'd be PROPOSAL__HOST, PROPOSAL__PORT etc. at the moment).
The way it's set up at the moment isn't ideal because it doesn't type check to access fields on the config. For now it means we'd have to store self.target_tables = config.target_tables in the initializer, rather than accessing them through the config later
2 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