Reputation: 11
I am trying to integrate netsuite using python. I am using oauth autentication and the credentials are working fine in postman. However I am getting the below error in python
HTTP request failed: 401 Client Error: Unauthorized for url
Code:
import requests
import time
import uuid
import hashlib
import hmac
import base64
from urllib.parse import quote
from requests_oauthlib import OAuth1Session
ACCOUNT_ID = 'YOUR_ACCOUNT_ID' # Your NetSuite account ID (e.g., 123456_SB1)
CONSUMER_KEY = 'YOUR_CONSUMER_KEY' # Consumer Key from Integration record
CONSUMER_SECRET = 'YOUR_CONSUMER_SECRET' # Consumer Secret from Integration record
TOKEN_ID = 'YOUR_TOKEN_ID' # Token ID from Access Token
TOKEN_SECRET = 'YOUR_TOKEN_SECRET' # Token Secret from Access Token
NETSUITE_BASE_URL = f"https://{ACCOUNT_ID.lower()}.suitetalk.api.netsuite.com"
# REST API Endpoint for Purchase Orders
PURCHASE_ORDERS_ENDPOINT = f"{NETSUITE_BASE_URL}/services/rest/record/v1/purchaseOrder"
def generate_oauth_header(url, method="GET"):
"""
Generates an OAuth 1.0 header for NetSuite API requests.
"""
# OAuth parameters
oauth_nonce = str(uuid.uuid4().hex)
oauth_timestamp = str(int(time.time()))
oauth_signature_method = "HMAC-SHA256"
oauth_version = "1.0"
# Base string for the signature
base_string = f"{method.upper()}&{quote(url, safe='')}&"
params = {
"realm": ACCOUNT_ID,
"oauth_consumer_key": CONSUMER_KEY,
"oauth_token": TOKEN_ID,
"oauth_nonce": oauth_nonce,
"oauth_timestamp": oauth_timestamp,
"oauth_signature_method": oauth_signature_method,
"oauth_version": oauth_version,
}
encoded_params = "&".join(f"{quote(k, safe='')}={quote(v, safe='')}" for k, v in sorted(params.items()))
base_string += quote(encoded_params, safe='')
print("\nBase String for Signature:")
print(base_string)
# Signing key
signing_key = f"{quote(CONSUMER_SECRET, safe='')}&{quote(TOKEN_SECRET, safe='')}"
# Generate signature
hashed = hmac.new(signing_key.encode('utf-8'), base_string.encode('utf-8'), hashlib.sha256)
oauth_signature = base64.b64encode(hashed.digest()).decode()
# Add signature to params
params["oauth_signature"] = oauth_signature
# Build the Authorization header
oauth_header = "OAuth " + ", ".join(f'{quote(k)}="{quote(v)}"' for k, v in params.items())
print("OAuth Authorization Header:")
print(oauth_header)
return oauth_header
def fetch_purchase_orders():
"""
Fetches a list of purchase orders from NetSuite using custom OAuth header.
"""
try:
# Generate OAuth header
oauth_header = generate_oauth_header(PURCHASE_ORDERS_ENDPOINT, method="GET")
headers = {
"Authorization": oauth_header,
"Content-Type": "application/json",
"Prefer": "transient" # Optional: For performance, avoid caching
}
# Make the GET request
response = requests.get(PURCHASE_ORDERS_ENDPOINT, headers=headers)
response.raise_for_status()
# Parse and print the response
data = response.json()
purchase_orders = data.get('items', [])
if not purchase_orders:
print("No purchase orders found.")
else:
for po in purchase_orders:
print(f"PO ID: {po['id']}, Vendor: {po.get('entity', {}).get('refName')}, Total: {po.get('total')}")
except requests.exceptions.RequestException as e:
print(f"HTTP request failed: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
if __name__ == "__main__":
fetch_purchase_orders()
Upvotes: 0
Views: 27