- Overview
- Platform setup and administration
- Platform setup and administration
- Platform architecture
- Data Bridge onboarding overview
- Connecting a Peak-managed data lake
- Connecting a customer-managed data lake
- Creating an AWS IAM role for Data Bridge
- Connecting a Snowflake data warehouse
- Connecting a Redshift data warehouse (public connectivity)
- Connecting a Redshift data warehouse (private connectivity)
- Reauthorizing a Snowflake OAuth connection
- Using Snowflake with Peak
- SQL Explorer overview
- Roles and permissions
- User management
- Inventory management solution
- Commercial pricing solution
- Merchandising solution

Supply Chain & Retail Solutions user guide
Uploading files via Python with signed URLs
Use Python to programmatically upload files to Peak via signed URLs. This approach is ideal for automating file uploads in data pipelines or scheduled batch processes.
Prerequisites
Before uploading files via Python, ensure you have the following:
- You have Python 3.6 or later installed.
- You have the
requestslibrary installed (pip install requests). - You have a tenant API key.
- You have the file ready for upload on your local machine.
Steps
Create a Python script to generate a signed URL and upload your file:
- Create a new Python script file (e.g.,
upload_to_peak.py). - Import the required libraries:
import os import requestsimport os import requests - Define the API key and file name. Store the API key in an environment variable instead of hardcoding it:
tenant_api_key = os.environ["peak_tenant_api_key"] file_name = "sample_20181112.csv"tenant_api_key = os.environ["peak_tenant_api_key"] file_name = "sample_20181112.csv" - Generate a signed URL:
url = "https://api.peak.ai/file/signedurl/client" headers = { "Content-Type": "application/json", "Authorization": tenant_api_key, } payload = {"fileName": file_name} response = requests.post(url, json=payload, headers=headers) response.raise_for_status() signed_url = response.text.strip('"')url = "https://api.peak.ai/file/signedurl/client" headers = { "Content-Type": "application/json", "Authorization": tenant_api_key, } payload = {"fileName": file_name} response = requests.post(url, json=payload, headers=headers) response.raise_for_status() signed_url = response.text.strip('"') - Open the file in binary mode and upload it using the signed URL:
with open(file_name, "rb") as file: upload_headers = {"Content-Type": "application/json"} put_response = requests.put(signed_url, data=file, headers=upload_headers) put_response.raise_for_status() print("File uploaded successfully")with open(file_name, "rb") as file: upload_headers = {"Content-Type": "application/json"} put_response = requests.put(signed_url, data=file, headers=upload_headers) put_response.raise_for_status() print("File uploaded successfully") - Save the script and run it:
python upload_to_peak.pypython upload_to_peak.py
Result
Your file uploads to Peak's file storage. The script outputs the upload status. The file becomes available for processing according to your configured data feeds and triggers.
Complete example script
Here is a complete Python 3 example that generates a signed URL and uploads a file:
import os
import requests
import sys
def upload_file_to_peak(file_name):
"""Generate a signed URL and upload a file to Peak."""
tenant_api_key = os.environ.get("peak_tenant_api_key")
if not tenant_api_key:
print("Missing environment variable: peak_tenant_api_key")
return False
try:
# Step 1: Generate signed URL.
signed_url_endpoint = "https://api.peak.ai/file/signedurl/client"
signed_url_headers = {
"Content-Type": "application/json",
"Authorization": tenant_api_key,
}
signed_url_payload = {"fileName": file_name}
print("Generating URL...")
first_response = requests.post(
signed_url_endpoint,
json=signed_url_payload,
headers=signed_url_headers,
)
if first_response.status_code != 200:
print(f"Signed URL request failed with status code: {first_response.status_code}")
print(f"Response: {first_response.text}")
return False
signed_url = first_response.text.strip('"')
# Step 2: Upload file with signed URL.
print("Uploading file...")
with open(file_name, "rb") as file_obj:
upload_headers = {"Content-Type": "application/json"}
put_response = requests.put(signed_url, data=file_obj, headers=upload_headers)
if put_response.status_code == 200:
print(f"File {file_name} uploaded successfully")
return True
print(f"Upload failed with status code: {put_response.status_code}")
print(f"Response: {put_response.text}")
return False
except FileNotFoundError:
print(f"File not found: {file_name}")
return False
except Exception as e:
print(f"Error during upload: {e}")
return False
# Usage
file_name = "sample_20181112.csv"
success = upload_file_to_peak(file_name)
if not success:
sys.exit(1)
import os
import requests
import sys
def upload_file_to_peak(file_name):
"""Generate a signed URL and upload a file to Peak."""
tenant_api_key = os.environ.get("peak_tenant_api_key")
if not tenant_api_key:
print("Missing environment variable: peak_tenant_api_key")
return False
try:
# Step 1: Generate signed URL.
signed_url_endpoint = "https://api.peak.ai/file/signedurl/client"
signed_url_headers = {
"Content-Type": "application/json",
"Authorization": tenant_api_key,
}
signed_url_payload = {"fileName": file_name}
print("Generating URL...")
first_response = requests.post(
signed_url_endpoint,
json=signed_url_payload,
headers=signed_url_headers,
)
if first_response.status_code != 200:
print(f"Signed URL request failed with status code: {first_response.status_code}")
print(f"Response: {first_response.text}")
return False
signed_url = first_response.text.strip('"')
# Step 2: Upload file with signed URL.
print("Uploading file...")
with open(file_name, "rb") as file_obj:
upload_headers = {"Content-Type": "application/json"}
put_response = requests.put(signed_url, data=file_obj, headers=upload_headers)
if put_response.status_code == 200:
print(f"File {file_name} uploaded successfully")
return True
print(f"Upload failed with status code: {put_response.status_code}")
print(f"Response: {put_response.text}")
return False
except FileNotFoundError:
print(f"File not found: {file_name}")
return False
except Exception as e:
print(f"Error during upload: {e}")
return False
# Usage
file_name = "sample_20181112.csv"
success = upload_file_to_peak(file_name)
if not success:
sys.exit(1)
Next steps
After the file is uploaded to the Peak data lake, create a feed and schedule it for recurring ingestion. For details, see Creating a file feed from Files.
Troubleshooting
If you encounter errors during upload, refer to the following common issues and solutions:
- 403 Forbidden - The signed URL may have expired. Request a new signed URL.
- 401 Unauthorized - Verify that the
Authorizationheader contains a valid tenant API key. - 400 Bad Request - Verify that the signed URL request body includes a valid
fileNamevalue. - FileNotFoundError - Check that the file path is correct and the file exists.
- Connection errors - Verify your network connection and firewall settings.