Most custom connectors extract from APIs:
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential())
def fetch_page(url, params):
response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
def extract_all(base_url):
results = []
page = 1
while True:
data = fetch_page(base_url, {'page': page, 'per_page': 100})
if not data['items']:
break
results.extend(data['items'])
page += 1
return results
Handle pagination, retries, and rate limits.