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

feat: Activate Version with example test #89

Merged
merged 13 commits into from
Feb 14, 2023
Merged

feat: Activate Version with example test #89

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
e25b99d
Activat Version example test
visch Jan 23, 2023
3280b01
Activate version working
visch Jan 23, 2023
43dc675
Lint and Schema fix for config
visch Jan 23, 2023
c433682
Activate Version test to show data being deleted
visch Jan 24, 2023
b367081
Lint Fix
visch Jan 24, 2023
f908796
Merge branch 'main' into activate_version
visch Jan 24, 2023
2ba215b
Activate version is doing things correctly, existing implementations …
visch Jan 24, 2023
fff8458
Merge branch 'activate_version' of github.com:MeltanoLabs/target-post…
visch Jan 24, 2023
67989b3
Merge branch 'main' into activate_version
visch Jan 30, 2023
e3fad47
Merge branch 'main' into activate_version
visch Jan 30, 2023
42441da
Merge branch 'main' into activate_version
visch Feb 1, 2023
f40ba63
Consistent spacing for docs
visch Feb 14, 2023
aacbed1
Update README.md
visch Feb 14, 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
30 changes: 16 additions & 14 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,20 +14,22 @@ Built with the [Meltano SDK](https://sdk.meltano.com) for Singer Taps and Target
* `schema-flattening`

## Settings

| Setting | Required | Default | Description |
| :------------------- | :------: | :-----------------: | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| host | False | None | Hostname for postgres instance. Note if sqlalchemy_url is set this will be ignored. |
| port | False | 5432 | The port on which postgres is awaiting connection. Note if sqlalchemy_url is set this will be ignored. |
| user | False | None | User name used to authenticate. Note if sqlalchemy_url is set this will be ignored. |
| password | False | None | Password used to authenticate. Note if sqlalchemy_url is set this will be ignored. |
| database | False | None | Database name. Note if sqlalchemy_url is set this will be ignored. |
| sqlalchemy_url | False | None | SQLAlchemy connection string. This will override using host, user, password, port, dialect. Note that you must esacpe password special characters properly see https://docs.sqlalchemy.org/en/20/core/engines.html#escaping-special-characters-such-as-signs-in-passwords |
| dialect+driver | False | postgresql+psycopg2 | Dialect+driver see https://docs.sqlalchemy.org/en/20/core/engines.html. Generally just leave this alone. Note if sqlalchemy_url is set this will be ignored. |
| stream_maps | False | None | Config object for stream maps capability. For more information check out [Stream Maps](https://sdk.meltano.com/en/latest/stream_maps.html). |
| stream_map_config | False | None | User-defined config values to be used within map expressions. |
| flattening_enabled | False | None | 'True' to enable schema flattening and automatically expand nested properties. |
| flattening_max_depth | False | None | The max depth to flatten schemas. |
| Setting | Required | Default | Description |
| :-------------------- | :------: | :-----------------: | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| host | False | None | Hostname for postgres instance. Note if sqlalchemy_url is set this will be ignored. |
| port | False | 5432 | The port on which postgres is awaiting connection. Note if sqlalchemy_url is set this will be ignored. |
| user | False | None | User name used to authenticate. Note if sqlalchemy_url is set this will be ignored. |
| password | False | None | Password used to authenticate. Note if sqlalchemy_url is set this will be ignored. |
| database | False | None | Database name. Note if sqlalchemy_url is set this will be ignored. |
| sqlalchemy_url | False | None | SQLAlchemy connection string. This will override using host, user, password, port,dialect. Note that you must esacpe password specialcharacters properly seehttps://docs.sqlalchemy.org/en/20/core/engines.html#escaping-special-characters-such-as-signs-in-passwords |
visch marked this conversation as resolved.
Show resolved Hide resolved
| dialect+driver | False | postgresql+psycopg2 | Dialect+driver see https://docs.sqlalchemy.org/en/20/core/engines.html. Generally just leave this alone. Note if sqlalchemy_url is set this will be ignored. |
| default_target_schema | False | None | Postgres schema to send data to, example: tap-clickup |
| hard_delete | False | 0 | When activate version is sent from a tap this specefies if we should delete the records that don't match, or mark them with a date in the `_sdc_deleted_at` column. |
Copy link
Member Author

Choose a reason for hiding this comment

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

Boolean should be generated as True / False here I think? I'll leave for now

Copy link
Contributor

Choose a reason for hiding this comment

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

Agreed. Very strange that it shows defaults of 0. Agreed though we don't need to block on this.

| add_record_metadata | False | 1 | Note that this must be enabled for activate_version to work!This adds _sdc_extracted_at, _sdc_batched_at, and more to every table. See https://sdk.meltano.com/en/latest/implementation/record_metadata.html for more information. |
| stream_maps | False | None | Config object for stream maps capability. For more information check out [Stream Maps](https://sdk.meltano.com/en/latest/stream_maps.html). |
| stream_map_config | False | None | User-defined config values to be used within map expressions. |
| flattening_enabled | False | None | 'True' to enable schema flattening and automatically expand nested properties. |
| flattening_max_depth | False | None | The max depth to flatten schemas. |

A full list of supported settings and capabilities is available by running: `target-postgres --about`

Expand Down
62 changes: 62 additions & 0 deletions target_postgres/sinks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,12 @@
import uuid
from typing import Any, Dict, Iterable, List, Optional, Union

import sqlalchemy
from pendulum import now
from singer_sdk.sinks import SQLSink
from sqlalchemy import Column, MetaData, Table, insert
from sqlalchemy.sql import Executable
from sqlalchemy.sql.expression import bindparam

from target_postgres.connector import PostgresConnector

Expand Down Expand Up @@ -252,3 +255,62 @@ def schema_name(self) -> Optional[str]:

# Schema name not detected.
return None

def activate_version(self, new_version: int) -> None:
Copy link
Member Author

Choose a reason for hiding this comment

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

I think we could take this function and just drop it into the SDK as it is here, this just has a few tweaks that are needed to make this work in a more generic way

"""Bump the active version of the target table.

Args:
new_version: The version number to activate.
"""
# There's nothing to do if the table doesn't exist yet
# (which it won't the first time the stream is processed)
if not self.connector.table_exists(self.full_table_name):
return

deleted_at = now()
datetime_type = self.connector.to_sql_type(
{"type": "string", "format": "date-time"}
) # Different from SingerSDK as we need to handle types the same as SCHEMA messsages
visch marked this conversation as resolved.
Show resolved Hide resolved
integer_type = self.connector.to_sql_type(
{"type": "integer"}
) # Different from SingerSDK as we need to handle types the same as SCHEMA messsages

if not self.connector.column_exists(
full_table_name=self.full_table_name,
column_name=self.version_column_name,
):
self.connector.prepare_column(
self.full_table_name,
self.version_column_name,
sql_type=integer_type, # Different from SingerSDK as we need to handle types the same as SCHEMA messsages
)

self.logger.info("Hard delete: %s", self.config.get("hard_delete"))
if self.config["hard_delete"] is True:
self.connection.execute(
f"DELETE FROM {self.full_table_name} "
visch marked this conversation as resolved.
Show resolved Hide resolved
f"WHERE {self.version_column_name} <= {new_version} OR {self.version_column_name} IS NULL"
Copy link
Member Author

Choose a reason for hiding this comment

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

I'm a little worried about what would happen if someone ran this in production with add_record_metadata set to True for a while, then suddenly they change it to False.

I think we'd end up dropping the entire table, because we'd add all the new records, then none of them would have a new_version set they'd all be NULL. Then we'd drop all the NULLs. :O

Removing the IS NULL check would work here but I think the better thing to do would be a check to be sure metadata_columns is enabled as I think the IS NULL check is very valid here as in normal operation you want activate_version to be keeping your data very "clean" and in the same shape as the source.

I should also write a test for this case as it's potentially scary (only place in our code that we drop tables so being cautious is good)

Copy link
Member Author

Choose a reason for hiding this comment

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

If we fix the other issue https://github.com/MeltanoLabs/target-postgres/pull/89/files#r1084618420 I believe this one gets solved as well.

)
return

if not self.connector.column_exists(
full_table_name=self.full_table_name,
column_name=self.soft_delete_column_name,
):
self.connector.prepare_column(
self.full_table_name,
self.soft_delete_column_name,
sql_type=datetime_type, # Different from SingerSDK as we need to handle types the same as SCHEMA messsages
)

query = sqlalchemy.text(
f"UPDATE {self.full_table_name}\n"
f"SET {self.soft_delete_column_name} = :deletedate \n"
f"WHERE {self.version_column_name} < :version OR {self.version_column_name} IS NULL \n" # Need to deal with the case where data doesn't exist for the version column
f" AND {self.soft_delete_column_name} IS NULL\n"
)
query = query.bindparams(
bindparam("deletedate", value=deleted_at, type_=datetime_type),
bindparam("version", value=new_version, type_=integer_type),
)
self.connector.connection.execute(query)
33 changes: 32 additions & 1 deletion target_postgres/target.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ def __init__(
th.StringType,
description=(
"User name used to authenticate. "
+ "Note if sqlalchemy_url is set this will be ignored.",
visch marked this conversation as resolved.
Show resolved Hide resolved
+ "Note if sqlalchemy_url is set this will be ignored."
),
),
th.Property(
Expand Down Expand Up @@ -117,6 +117,37 @@ def __init__(
th.StringType,
description="Postgres schema to send data to, example: tap-clickup",
),
th.Property(
"hard_delete",
th.StringType,
default=False,
description=(
"When activate version is sent from a tap this specefies "
+ "if we should delete the records that don't match, or mark "
+ "them with a date in the `_sdc_deleted_at` column."
),
),
th.Property(
"hard_delete",
th.StringType,
default=False,
description=(
"When activate version is sent from a tap this specefies "
+ "if we should delete the records that don't match, or mark "
+ "them with a date in the `_sdc_deleted_at` column."
),
),
th.Property(
"add_record_metadata",
th.StringType,
default=True,
description=(
"Note that this must be enabled for activate_version to work!"
+ "This adds _sdc_extracted_at, _sdc_batched_at, and more to every table. "
+ "See https://sdk.meltano.com/en/latest/implementation/record_metadata.html "
+ "for more information."
),
),
).to_dict()
default_sink_class = PostgresSink

Expand Down
10 changes: 10 additions & 0 deletions target_postgres/tests/data_files/activate_version_hard.singer
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
{"type": "SCHEMA", "stream": "test_activate_version_hard", "schema": {"type": "object", "properties": {"code": {"type": ["string"]}, "name": {"type": ["null", "string"]}}}, "key_properties": ["code"], "bookmark_properties": []}
{"type": "ACTIVATE_VERSION", "stream": "test_activate_version_hard", "version": 1674486431563}
{"type": "RECORD", "stream": "test_activate_version_hard", "record": {"code": "AF", "name": "Africa"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_hard", "record": {"code": "AN", "name": "Antarctica"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_hard", "record": {"code": "AS", "name": "Asia"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_hard", "record": {"code": "EU", "name": "Europe"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_hard", "record": {"code": "NA", "name": "North America"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_hard", "record": {"code": "OC", "name": "Oceania"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_hard", "record": {"code": "SA", "name": "South America"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "ACTIVATE_VERSION", "stream": "test_activate_version_hard", "version": 1674486431563}
10 changes: 10 additions & 0 deletions target_postgres/tests/data_files/activate_version_soft.singer
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
{"type": "SCHEMA", "stream": "test_activate_version_soft", "schema": {"type": "object", "properties": {"code": {"type": ["string"]}, "name": {"type": ["null", "string"]}}}, "key_properties": ["code"], "bookmark_properties": []}
{"type": "ACTIVATE_VERSION", "stream": "test_activate_version_soft", "version": 1674486431563}
{"type": "RECORD", "stream": "test_activate_version_soft", "record": {"code": "AF", "name": "Africa"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_soft", "record": {"code": "AN", "name": "Antarctica"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_soft", "record": {"code": "AS", "name": "Asia"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_soft", "record": {"code": "EU", "name": "Europe"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_soft", "record": {"code": "NA", "name": "North America"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_soft", "record": {"code": "OC", "name": "Oceania"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "RECORD", "stream": "test_activate_version_soft", "record": {"code": "SA", "name": "South America"}, "version": 1674486431563, "time_extracted": "2023-01-23T15:07:11.563063Z"}
{"type": "ACTIVATE_VERSION", "stream": "test_activate_version_soft", "version": 1674486431563}
Loading