FedEx RESTful API
zesriver opened this issue · 4 comments
zesriver commented
Hello,
Thank you for your help.
I received the following email from FedEx.
"Migrate to the FedEx RESTful API by February 28, 2024 before FedEx Web Services is retired."
Do you have any plans to release a FedEx RESTful API version?
MatinF commented
We would be very interested in this as well
MatinF commented
Has anybody found examples of 'full examples' of the REST API being used with Python e.g. to create international shipments?
MatinF commented
I created a class for use in creating international shipments. It's setup assuming that the origin country is Denmark, but that part can easily be modified. I hope some of you will find this useful - the code will need to be adapted for your use case.
FedEx helper class
class FedexLabelHelper:
def __init__(self, inp, mode):
self.inp = inp
self.fedex_cred = inp.fedex_cred[mode]
self.mode = mode
self.access_token = ""
return
# initialize session by creating access_token for use in other requests
def get_access_token(self):
import json, requests
url = self.fedex_cred["url_prefix"] + "/oauth/token"
payload = {
"grant_type": "client_credentials",
"client_id": self.fedex_cred["key"],
"client_secret": self.fedex_cred["password"],
}
headers = {"Content-Type": "application/x-www-form-urlencoded"}
response = requests.request("POST", url, data=payload, headers=headers)
access_token = json.loads(response.text)["access_token"]
self.access_token = access_token
# specify recipient details
def set_recipients(self, recipient):
if recipient["CountryCode"] == "US" or recipient["CountryCode"] == "CA":
region = recipient["Region"]
else:
region = ""
payload_recipients = [
{
"contact": {
"personName": recipient["Name"],
"phoneNumber": recipient["Phone"],
},
"address": {
"streetLines": recipient["Address"],
"city": recipient["City"],
"stateOrProvinceCode": region,
"postalCode": recipient["Zip"],
"countryCode": recipient["CountryCode"],
"residential": recipient["Residential"],
},
}
]
return payload_recipients
# set email notification details
def set_notifications(self, recipient):
email = recipient["Email"]
if self.inp.config["send_to_self"] == True or self.mode != "production":
email = self.inp.shipper["contact"]["emailAddress"]
payload_notifications = {
"aggregationType": "PER_PACKAGE",
"emailNotificationRecipients": [
{
"name": recipient["Name"],
"emailNotificationRecipientType": "RECIPIENT",
"emailAddress": email,
"notificationFormatType": "TEXT",
"notificationType": "EMAIL",
"locale": "en_US",
"notificationEventType": self.inp.config["notificationEventType"],
}
],
"personalMessage": "",
}
return payload_notifications
# set PDF label details
def set_label_specs(self):
payload_label_specs = {
"imageType": self.inp.config["imageType"],
"labelStockType": self.inp.config["labelStockType"],
"labelFormatType": self.inp.config["labelFormatType"],
"labelPrintingOrientation": self.inp.config["labelPrintingOrientation"],
"LabelOrder": self.inp.config["LabelOrder"],
}
return payload_label_specs
# specify customs clearance details
def set_customs(self, invoice_info):
unit_price = float(round((invoice_info["Value"] / invoice_info["Quantity"]), 2))
payload_customs = {
"totalCustomsValue": {
"amount": invoice_info["Value"],
"currency": invoice_info["Currency"],
},
"dutiesPayment": {
"paymentType": "RECIPIENT",
},
"commodities": [
{
"description": "See attached commercial invoice",
"countryOfManufacture": "DK",
"numberOfPieces": invoice_info["Quantity"],
"weight": {"value": invoice_info["FinalWeight"], "units": "KG"},
"quantity": invoice_info["Quantity"],
"quantityUnits": "EA",
"unitPrice": {
"amount": unit_price,
"currency": invoice_info["Currency"],
},
"customsValue": {
"amount": invoice_info["Value"],
"currency": invoice_info["Currency"],
},
"exportLicenseNumber": self.inp.shipper["tins"][0]["number"],
}
],
}
return payload_customs
# specify packaging type details
def set_package_line_items(self, invoice_info):
payload_package_line_items = [
{
"physicalPackaging": invoice_info["PackagingType"],
"weight": {"units": "KG", "value": invoice_info["FinalWeight"]},
}
]
return payload_package_line_items
# set general information incl. service type
def set_general_info(self, payload, invoice_info):
from datetime import datetime
# determine service_type
if (
payload["requestedShipment"]["recipients"][0]["address"]["countryCode"]
== "DK"
):
service_type = "PRIORITY_OVERNIGHT"
else:
service_type = (
"INTERNATIONAL_PRIORITY"
if invoice_info["ShippingExpress"] == True
else "INTERNATIONAL_ECONOMY"
)
str_today = datetime.today().strftime("%Y-%m-%d")
payload["requestedShipment"]["shipDatestamp"] = str_today
payload["requestedShipment"]["pickupType"] = self.inp.config["pickupType"]
payload["requestedShipment"]["serviceType"] = service_type
payload["requestedShipment"]["packagingType"] = invoice_info["PackagingType"]
payload["requestedShipment"]["blockInsightVisibility"] = False
payload["requestedShipment"]["shippingChargesPayment"] = {
"paymentType": "SENDER"
}
payload["accountNumber"] = {"value": self.fedex_cred["freight_account_number"]}
payload["labelResponseOptions"] = "URL_ONLY"
return payload
# assign doc_ids from uploaded documents and enable notifications
def set_special_services(self, doc_ids):
attached_documents = []
for i, doc_id in enumerate(doc_ids, start=0):
doc = {}
doc["documentType"] = "COMMERCIAL_INVOICE" if i == 0 else "OTHER"
doc["documentReference"] = "DocumentReference"
doc["description"] = (
"COMMERCIAL_INVOICE" if i == 0 else "Product_Description"
)
doc["documentId"] = doc_id
attached_documents.append(doc)
payload_special_services = {
"specialServiceTypes": ["ELECTRONIC_TRADE_DOCUMENTS", "EVENT_NOTIFICATION"],
"etdDetail": {"attachedDocuments": attached_documents},
}
return payload_special_services
# validate shipment (NOT USED)
def validate_shipment(self, payload):
import json, requests
url = self.fedex_cred["url_prefix"] + "/ship/v1/shipments/packages/validate"
payload = json.dumps(payload)
headers = {
"Content-Type": "application/json",
"X-locale": "en_DK",
"Authorization": f"Bearer {self.access_token}",
}
response = requests.request("POST", url, data=payload, headers=headers)
if response.status_code == 200:
print("- OK: Validated shipment\n")
else:
print("- WARNING: Unable to validate shipment\n")
print(response.text)
# create shipment and store PDF invoices/labels in output folder
def create_shipment(self, payload, pdfs, output_path_date, invoice_info, recipient):
import json, requests
from PyPDF2 import PdfReader, PdfWriter
from shutil import copyfile
tracking_id = ""
recipient_email = recipient["Email"]
if recipient_email == "":
recipient_email = "No email - add tracking in Amazon Seller Central"
url = self.fedex_cred["url_prefix"] + "/ship/v1/shipments"
payload = json.dumps(payload)
headers = {
"Content-Type": "application/json",
"X-locale": "en_DK",
"Authorization": f"Bearer {self.access_token}",
}
response = requests.request("POST", url, data=payload, headers=headers)
if response.status_code != 200:
print("- WARNING: Unable to process shipment\n")
print("recipient: ", recipient)
print(response.text)
return pdfs, tracking_id
else:
# extract information from the create shipment response
response_json = json.loads(response.text)
shipment = response_json["output"]["transactionShipments"][0]
tracking_id = shipment["masterTrackingNumber"]
label_url = shipment["pieceResponses"][0]["packageDocuments"][0]["url"]
# download the label PDF, open it, get the 1st page and save that only
output_path = output_path_date / (
invoice_info["InvoiceId"] + f"_shipment_label_{tracking_id}.pdf"
)
label_response = requests.get(label_url)
with open(output_path, "wb") as f:
f.write(label_response.content)
f.close()
with open(output_path, "rb") as in_pdf:
reader = PdfReader(in_pdf)
writer = PdfWriter()
writer.add_page(reader.pages[0])
with open(output_path, "wb") as out_pdf:
writer.write(out_pdf)
out_pdf.close()
# copy the cleaned invoice PDF into the final destination folder
if self.mode == "production":
copyfile(
invoice_info["InvoicePath"],
output_path_date / invoice_info["InvoicePath"].name,
)
# print summary
if self.mode == "production":
print(
f"- OK: Created shipment | tracking: {tracking_id} | email: {recipient_email}\n"
)
else:
print(f"- OK: Validated shipment\n")
# add path to invoice and label for PDF list (for subsequent merge)
if self.mode == "production":
pdfs.append(invoice_info["InvoicePath"]) # path to invoice
pdfs.append(output_path) # path to label
return pdfs, tracking_id
# create pickup if it does not already exist
def create_pickup(self, root, pack_weight, pack_count):
import json, requests, datetime, os
# check if a folder already exists for today
subdirs = [
x[0].split("\\")[-1][0:8]
for x in os.walk(root / self.inp.paths["output"][self.mode])
]
date_now = datetime.datetime.now().strftime("%Y%m%d")
if date_now in subdirs[-2]:
print("Output folder already exists for today - no pickup scheduled")
return
else:
pickup_location = {}
pickup_location["contact"] = self.inp.shipper["contact"]
pickup_location["address"] = self.inp.shipper["address"]
payload = {
"associatedAccountNumber": {
"value": self.fedex_cred["freight_account_number"]
},
"originDetail": {
"pickupLocation": pickup_location,
"readyDateTimestamp": self.inp.pickup_date,
"customerCloseTime": self.inp.config["customerCloseTime"],
},
"countryRelationships": "INTERNATIONAL",
"carrierCode": "FDXE",
"totalWeight": {"units": "KG", "value": pack_weight},
"PackageCount": pack_count,
}
url = self.fedex_cred["url_prefix"] + "/pickup/v1/pickups"
payload = json.dumps(payload)
headers = {
"Content-Type": "application/json",
"X-locale": "en_DK",
"Authorization": f"Bearer {self.access_token}",
}
response = requests.request("POST", url, data=payload, headers=headers)
if response.status_code != 200:
print(f"\n\nWARNING: Failed to create pickup\n")
print(response.text)
else:
confirmation_code = json.loads(response.text)["output"][
"pickupConfirmationCode"
]
print(f"\n\nOK: Pickup booked with code {confirmation_code}\n")
return confirmation_code
# upload both the commercial invoice and the combined product descriptions PDF
def upload_all_documents(self, root, recipient, invoice_info):
doc_ids = []
# get doc_id for commercial invoice
doc_ids.append(
self.upload_document(
invoice_info["InvoicePath"], "COMMERCIAL_INVOICE", recipient
)
)
# get doc_id for product descriptions
doc_ids.append(
self.upload_document(
root / self.inp.paths["product_descriptions"],
"OTHER",
recipient,
)
)
return doc_ids
# function for uploading PDF file from local disk
def upload_document(self, path, type, recipient):
import requests, binascii, json
from requests_toolbelt import MultipartEncoder
file_name = path.name
country_code = recipient["CountryCode"]
file_content = open(path, "rb").read()
url = self.fedex_cred["doc_url_prefix"] + "/documents/v1/etds/upload"
payload = {
"document": f'{{"workflowName": "ETDPreshipment", "carrierCode": "FDXE", "name": "{file_name}", "contentType": "application/pdf", "meta": {{"shipDocumentType": "{type}", "originCountryCode": "DK", "destinationCountryCode": "{country_code}"}}}}',
"attachment": (file_name, file_content, "application/pdf"),
}
m = MultipartEncoder(fields=payload)
headers = {
"Authorization": f"Bearer {self.access_token}",
"x-customer-transaction-id": "ETD-Pre-Shipment-Upload_test1",
"Content-Type": m.content_type,
}
response = requests.post(url, headers=headers, data=m)
if response.status_code == 201:
response_json = json.loads(response.text)
print(response_json)
return response_json["output"]["meta"]["docId"]
else:
print("- WARNING: Unable to upload document:", response)
return ""
inputs.py file for providing inputs:
import datetime
# FedEx test credentails
fedex_cred = {
"draft": {
"url_prefix": "https://apis-sandbox.fedex.com",
"doc_url_prefix": "https://documentapitest.prod.fedex.com/sandbox",
"key": "REDACTED",
"password": "REDACTED",
"freight_account_number": "REDACTED",
},
"production": {
"url_prefix": "https://apis.fedex.com",
"doc_url_prefix": "https://documentapi.prod.fedex.com",
"key": "REDACTED",
"password": "REDACTED",
"freight_account_number": "REDACTED",
},
}
fedex_cred["test"] = fedex_cred["draft"]
# shipper information
shipper = {
"address": {
"city": "CITY",
"stateOrProvinceCode": "",
"postalCode": "ZIP",
"countryCode": "DK",
"residential": False,
"streetLines": ["ADDRESS"],
},
"contact": {
"personName": "Your Name",
"emailAddress": "mail@mail.com",
"phoneExtension": "45",
"phoneNumber": "12345678",
"companyName": "Your Company",
},
"tins": [
{
"number": "DK12345678",
"tinType": "BUSINESS_NATIONAL",
"usage": "usage",
"effectiveDate": "2000-01-23T04:56:07.000+00:00",
"expirationDate": "2040-01-23T04:56:07.000+00:00",
}
],
}
# other configuration details
config = {
"send_to_self": False,
"domestic": False, # toggle whether to enable domestic shipments (WIP)
"isDebug": False, # toggle wheter to provide debug printout
"cust_tran_id": "Something - Automated Label Creation", # custom ID sent in FedEx requests
"shippingPaymentType": "SENDER", # control who pays freight charges
"labelFormatType": "COMMON2D", # label details
"imageType": "PDF", # label details
"labelStockType": "PAPER_7X475", # label details
"labelPrintingOrientation": "TOP_EDGE_OF_TEXT_FIRST", # label details
"LabelOrder": "SHIPPING_LABEL_FIRST", # label details
"BoxLargeWeight": 0.75,
"BoxMediumWeight": 0.55,
"BoxMediumSmallWeight": 0.45,
"BoxSmallWeight": 0.25,
"pickupType": "USE_SCHEDULED_PICKUP",
"notificationEventType": [
"ON_DELIVERY",
"ON_EXCEPTION",
"ON_PICKUP_DRIVER_ARRIVED",
],
"customerCloseTime": "16:00:00",
"shopify_tag": "Order ID: #"
}
# specify paths and load data
paths = {
"invoices_raw": "invoices_raw/",
"invoices": "invoices/",
"descr_folder": "product-descriptions/",
"output": {
"test": "fedexlabels_test/",
"production": "fedexlabels_production/",
"draft": "fedexlabels_draft/",
},
"country_codes": "data-tables/country-codes.csv",
"commodity_data": "data-tables/commodity-data.csv",
"merged_pdf": "all_shipments.pdf",
"product_descriptions": "data-tables/product-descriptions_compressed.pdf"
}
# UTC pickup time
pickup_date = (
datetime.datetime.today()
.replace(hour=12, minute=0, second=0)
.replace(microsecond=0)
).isoformat() + "Z"
main.py for running the script
from pathlib import Path
import json
import inputs as inp
from pathlib import Path
from utils_fedex import FedexLabelHelper
from utils_rest import (
EconomicHelper,
ShopifyHelper,
get_country_codes,
get_recipient_info,
get_invoice_info,
clean_recipient,
check_exceptions,
create_output_path_date,
merge_pdf_files,
store_manual_invoices,
print_invoice_info,
)
mode = "draft" # *** TEMPORARY ***
# initialize e-conomic, Shopify and FedEx classes
eco = EconomicHelper(inp, mode)
sho = ShopifyHelper(inp, mode)
flh = FedexLabelHelper(inp, mode)
flh.get_access_token()
# initialize variables
root = Path(__file__).parent
pdfs = []
pack_weight = 0
pack_count = 0
country_codes = get_country_codes(root / inp.paths["country_codes"])
output_path_date = create_output_path_date(root / inp.paths["output"][mode])
shopify_orders = sho.get_shopify_orders()
# get invoice IDs and data from e-conomic based on mode
invoice_ids, invoice_data = eco.get_invoice_ids_data(root)
# loop through draft invoices or downloaded invoices
for i, invoice in enumerate(invoice_data, start=1):
invoice_id = str(invoice_ids[i - 1])
# --- 1: EXTRACT INVOICE DETAILS FROM YOUR ACCOUNTING SYSTEM ---
# get recipient shipping details
recipient = get_recipient_info(invoice, root, inp, mode)
invoice_info = get_invoice_info(invoice, root, inp, mode)
# --- 2: CREATE FEDEX LABEL ---
# construct the shipment JSON payload
payload = {"requestedShipment": {}}
payload_req = payload["requestedShipment"]
payload_req["shipper"] = inp.shipper
payload_req["recipients"] = flh.set_recipients(recipient)
payload_req["emailNotificationDetail"] = flh.set_notifications(recipient)
payload_req["customsClearanceDetail"] = flh.set_customs(invoice_info)
payload_req["labelSpecification"] = flh.set_label_specs()
payload_req["requestedPackageLineItems"] = flh.set_package_line_items(invoice_info)
payload = flh.set_general_info(payload, invoice_info)
# if production mode, upload PDF documents and as as ETD
if mode == "production":
doc_ids = flh.upload_all_documents(root, recipient, invoice_info)
payload_req["shipmentSpecialServices"] = flh.set_special_services(doc_ids)
pdfs, tracking_id = flh.create_shipment(
payload, pdfs, output_path_date, invoice_info, recipient
)
# update pickup details
pack_count += 1
pack_weight += invoice_info["Weight"]
# flh.create_pickup(root, pack_weight, pack_count)
radzhome commented
I don't use this package so I dont have time to maintain it. If you need me to merge anything I can look but need to see a PR.