Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: log based replication #249

Merged
merged 20 commits into from
Oct 16, 2023
Merged

feat: log based replication #249

Changes from 1 commit
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
5448d84
Initial setup for PostgresLogBasedStream
sebastianswms Sep 25, 2023
2be8274
Merge branch 'main' into log-based
sebastianswms Sep 25, 2023
03a2d6b
State storage
sebastianswms Sep 26, 2023
5232703
README updates and linting
sebastianswms Oct 10, 2023
5f53271
linting fix
sebastianswms Oct 10, 2023
6954d73
Merge branch 'main' into log-based
sebastianswms Oct 10, 2023
0e9782a
linting fix
sebastianswms Oct 10, 2023
aa40fcd
linting fix
sebastianswms Oct 10, 2023
2e37088
Update tap_postgres/tap.py
visch Oct 13, 2023
10e428f
Update README.md
visch Oct 13, 2023
be50571
Apply suggestions from code review
visch Oct 13, 2023
41a61e1
Update README.md
visch Oct 13, 2023
a1a136b
Update tap_postgres/client.py
visch Oct 13, 2023
c5f4548
Update tap_postgres/client.py
visch Oct 13, 2023
ea80c0d
WIP
sebastianswms Oct 16, 2023
386d37c
Merge branch 'log-based' of https://github.com/sebastianswms/tap-post…
sebastianswms Oct 16, 2023
c84dc21
Feature complete updates.
sebastianswms Oct 16, 2023
3db019c
linting with ruff and black
sebastianswms Oct 16, 2023
e95263d
more linting and slight refactor
sebastianswms Oct 16, 2023
8692033
Merge branch 'main' into log-based
visch Oct 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
linting with ruff and black
  • Loading branch information
@sebastianswms
sebastianswms committed Oct 16, 2023
commit 3db019ca3f52a7a233a690aa03cc8b385974a7a8
24 changes: 20 additions & 4 deletions tap_postgres/tap.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,11 @@
import paramiko
from singer_sdk import SQLTap, Stream
from singer_sdk import typing as th
from singer_sdk._singerlib import Catalog, Metadata, Schema # JSON schema typing helpers
from singer_sdk._singerlib import ( # JSON schema typing helpers
Catalog,
Metadata,
Schema,
)
from sqlalchemy.engine import URL
from sqlalchemy.engine.url import make_url
from sshtunnel import SSHTunnelForwarder
Expand Down Expand Up @@ -513,7 +517,7 @@ def catalog_dict(self) -> dict:
@property
def catalog(self) -> Catalog:
return self.modify_catalog(super().catalog)

def modify_catalog(self, catalog: Catalog) -> dict:
new_catalog: Catalog = Catalog()
modified_streams: list = []
Expand All @@ -533,13 +537,25 @@ def modify_catalog(self, catalog: Catalog) -> dict:
new_stream.schema.properties.update(
{"_sdc_deleted_at": Schema(type=["string", "null"])}
)
new_stream.metadata.update({("properties", "_sdc_deleted_at"): Metadata(Metadata.InclusionType.AVAILABLE, True, None)})
new_stream.metadata.update(
{
("properties", "_sdc_deleted_at"): Metadata(
Metadata.InclusionType.AVAILABLE, True, None
)
}
)
if "_sdc_lsn" not in new_stream.schema.properties:
stream_modified = True
new_stream.schema.properties.update(
{"_sdc_lsn": Schema(type=["integer", "null"])}
)
new_stream.metadata.update({("properties", "_sdc_lsn"): Metadata(Metadata.InclusionType.AVAILABLE, True, None)})
new_stream.metadata.update(
{
("properties", "_sdc_lsn"): Metadata(
Metadata.InclusionType.AVAILABLE, True, None
)
}
)
if stream_modified:
modified_streams.append(new_stream.tap_stream_id)
new_catalog.add_stream(new_stream)
Expand Down
Loading