Skip to content

fix: correct ShopifyQL schema discovery and timestamp handling #153

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

Open
naveen-meltano wants to merge 2 commits into
base: master
Choose a base branch
from
Open

fix: correct ShopifyQL schema discovery and timestamp handling #153

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6ddbc82
fix: correct ShopifyQL schema discovery and timestamp handling
naveen-meltano Jul 3, 2026
b6d8fa8
style: apply black formatting
naveen-meltano Jul 3, 2026
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
48 changes: 40 additions & 8 deletions tap_shopify/streams.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -379,16 +379,23 @@ def __init__(self, *args, **kwargs):
query_entry = kwargs.pop("query")
self.name = query_entry["name"]
self._configured_query = query_entry["query"]
self.primary_keys = query_entry.get("primary_keys") or []
self.replication_key = query_entry.get("replication_key")

super().__init__(*args, **kwargs)

# Must be set after super().__init__(), which resets
# self._primary_keys / self._replication_key to their defaults.
self.primary_keys = query_entry.get("primary_keys") or []
self.replication_key = query_entry.get("replication_key")

@cached_property
def schema(self) -> dict:
"""Discover schema by probing the API with a 1-day window."""
probe_query = re.sub(r"\bSINCE\s+\S+", "", self._configured_query, flags=re.IGNORECASE)
probe_query = re.sub(r"\bUNTIL\s+\S+", "", probe_query, flags=re.IGNORECASE).strip()
probe_query = re.sub(
r"\bSINCE\s+\S+", "", self._configured_query, flags=re.IGNORECASE
)
probe_query = re.sub(
r"\bUNTIL\s+\S+", "", probe_query, flags=re.IGNORECASE
).strip()
probe_query += " SINCE -1d UNTIL -0d"

graphql = self._GRAPHQL_TEMPLATE.format(shopifyql=json.dumps(probe_query))
Expand All @@ -400,15 +407,31 @@ def schema(self) -> dict:
response.raise_for_status()
data = response.json()

result = data.get("shopifyqlQuery") or {}
gql_errors = data.get("errors")
if gql_errors:
raise RuntimeError(
f"GraphQL errors during schema discovery for '{self.name}': {gql_errors}"
)

result = (data.get("data") or {}).get("shopifyqlQuery") or {}
parse_errors = result.get("parseErrors") or []
if parse_errors:
raise RuntimeError(
f"ShopifyQL parse error during schema discovery for '{self.name}': {parse_errors}"
)

columns = (result.get("tableData") or {}).get("columns") or []
props = [th.Property(col["name"], th.StringType) for col in columns]
props = [
th.Property(
col["name"],
(
th.DateTimeType
if col["name"] == self.replication_key
else th.StringType
),
)
for col in columns
]
return th.PropertiesList(*props).to_dict()

def prepare_request_payload(self, context, next_page_token):
Expand Down Expand Up @@ -441,7 +464,7 @@ def parse_response(self, response):
if gql_errors:
raise RuntimeError(f"GraphQL errors: {gql_errors}")

result = data.get("shopifyqlQuery") or {}
result = (data.get("data") or {}).get("shopifyqlQuery") or {}

parse_errors = result.get("parseErrors") or []
if parse_errors:
Expand All @@ -462,5 +485,14 @@ def get_url_params(self, context, next_page_token):
return {}

def post_process(self, row, context=None):
"""Return the row as-is — deduplication is not applicable here."""
"""Normalize the replication key to a full timestamp.

ShopifyQL returns date-only strings (e.g. "2026-07-03") for TIMESERIES
columns, but the declared date-time schema type — and BigQuery's
TIMESTAMP column type on load — require a full timestamp value.
"""
if self.replication_key and self.replication_key in row:
value = row[self.replication_key]
if isinstance(value, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
row[self.replication_key] = f"{value}T00:00:00Z"
return row
Loading