Skip to content

Commit

Permalink
Remove redundant JSON schema type reinforcement method in KlaviyoStre…
Browse files Browse the repository at this point in the history
…am to streamline post-processing. This change simplifies the code and enhances maintainability by relying on existing type handling mechanisms.
  • Loading branch information
butkeraites-hotglue committed Dec 2, 2024
1 parent 18bd56f commit c35abb0
Showing 1 changed file with 1 addition and 39 deletions.
40 changes: 1 addition & 39 deletions tap_klaviyo/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,50 +93,12 @@ def get_url_params(
params["filter"] = f"greater-than({self.replication_key},{start_date})"
return params

def _reinforce_jsonschema_type(self, obj, sub_schema):
if "null" in sub_schema["type"] and obj is None:
return None
if "integer" in sub_schema["type"]:
try:
return int(obj)
except:
pass
if "number" in sub_schema["type"]:
try:
return float(obj)
except:
pass
if "boolean" in sub_schema["type"]:
try:
return bool(obj)
except:
pass
if "string" in sub_schema["type"]:
if self.is_unix_timestamp(obj):
return obj
try:
return str(obj)
except:
pass
if type(obj) == list:
if len(obj) > 0:
return [self._reinforce_jsonschema_type(item, sub_schema["items"]) for item in obj]
else:
return []
if type(obj) == dict:
reinforced_obj = {}
for key in obj.keys():
reinforced_obj[key] = self._reinforce_jsonschema_type(obj[key], sub_schema["properties"][key])
return reinforced_obj
else:
raise Exception(f"Unsupported type: {type(obj)}")


def post_process(self, row, context):
row = super().post_process(row, context)
for key, value in row.get("attributes", {}).items():
row[key] = value
row.pop("attributes", None)
row = self._reinforce_jsonschema_type(row, self.schema)
return row

def is_unix_timestamp(self, date):
Expand Down

0 comments on commit c35abb0

Please sign in to comment.