Forum Discussion
AstridM
1 year agoAdvocate I
Rest api call with multiple pages
Hello, I need to migrate an SSIS package that make API calls and return multiple pages. The call goes to https://{{HOSTNAME}}/personnel/v1/person-details, and I can manually do https://{{HOSTNAME}}...
- 1 year ago
good morning, i used at the end a notebook.
something similar to:
import requests import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # API Configuration API_URL = "https://service3.ultipro.ca/personnel/v1/person-details?page={}" USERNAME = "YOUR_USERNAME" PASSWORD = "YOUR_PASSWORD" HEADERS = { "Authorization": "Bearer YOUR_API_TOKEN" } # Configure session with retries session = requests.Session() retry_strategy = Retry( total=5, status_forcelist=[500, 502, 503, 504], backoff_factor=2 # Exponential backoff ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("https://", adapter) # Pagination Variables page_number = 1 has_more_data = True all_data = [] while has_more_data: try: response = session.get( API_URL.format(page_number), auth=(USERNAME, PASSWORD), headers=HEADERS, timeout=10 # Set timeout to avoid hanging requests ) if response.status_code == 200: data = response.json() if not data: has_more_data = False break all_data.extend(data) page_number += 1 # Move to the next page elif response.status_code == 429: print("Rate limit hit, retrying after delay...") time.sleep(5) continue else: print(f"Error {response.status_code}: {response.text}") break except requests.exceptions.Timeout: print("Request timed out. Retrying...") time.sleep(5) # Wait before retrying except requests.exceptions.RequestException as e: print(f"Request failed: {e}") break print(f"Total records fetched: {len(all_data)}")
AstridM
1 year agoAdvocate I
good morning, i used at the end a notebook.
something similar to:
import requests
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# API Configuration
API_URL = "https://service3.ultipro.ca/personnel/v1/person-details?page={}"
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
HEADERS = {
"Authorization": "Bearer YOUR_API_TOKEN"
}
# Configure session with retries
session = requests.Session()
retry_strategy = Retry(
total=5,
status_forcelist=[500, 502, 503, 504],
backoff_factor=2 # Exponential backoff
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
# Pagination Variables
page_number = 1
has_more_data = True
all_data = []
while has_more_data:
try:
response = session.get(
API_URL.format(page_number),
auth=(USERNAME, PASSWORD),
headers=HEADERS,
timeout=10 # Set timeout to avoid hanging requests
)
if response.status_code == 200:
data = response.json()
if not data:
has_more_data = False
break
all_data.extend(data)
page_number += 1 # Move to the next page
elif response.status_code == 429:
print("Rate limit hit, retrying after delay...")
time.sleep(5)
continue
else:
print(f"Error {response.status_code}: {response.text}")
break
except requests.exceptions.Timeout:
print("Request timed out. Retrying...")
time.sleep(5) # Wait before retrying
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
break
print(f"Total records fetched: {len(all_data)}")