yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-08-08 09:18:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

sync improvements

Browse Source
This commit is contained in:
Jonny Dixon 2025-03-17 15:23:04 +00:00
parent 83afc6b3a5
commit 04b58753f7
1 changed files with 16 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
metadata-ingestion/src/datahub/ingestion/source/fivetran/fivetran_api_client.py
View File

@ -61,9 +61,20 @@ class FivetranAPIClient:
) )
kwargs["headers"] = headers kwargs["headers"] = headers
try:
response = self._session.request(method, url, **kwargs) response = self._session.request(method, url, **kwargs)
response.raise_for_status() response.raise_for_status()
return response.json() return response.json()
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP error occurred: {e}")
if e.response.status_code == 404:
logger.error(f"Endpoint not found: {url}")
if "sync_history" in endpoint:
logger.info("Attempting fallback to /sync endpoint")
# Try with the correct endpoint
new_endpoint = endpoint.replace("sync_history", "sync")
return self._make_request(method, new_endpoint, **kwargs)
raise
def list_connectors(self) -> List[Dict]: def list_connectors(self) -> List[Dict]:
"""Retrieve all connectors from the Fivetran API.""" """Retrieve all connectors from the Fivetran API."""
@ -172,8 +183,9 @@ class FivetranAPIClient:
since_time = int(time.time()) - (days * 24 * 60 * 60) since_time = int(time.time()) - (days * 24 * 60 * 60)
params = {"limit": 100, "since": since_time} params = {"limit": 100, "since": since_time}
# Updated to use the correct endpoint
response = self._make_request( response = self._make_request(
"GET", f"/connectors/{connector_id}/sync_history", params=params "GET", f"/connectors/{connector_id}/sync", params=params
) )
return response.get("data", {}).get("items", []) return response.get("data", {}).get("items", [])