python-fedex-devs / python-fedex

A light wrapper around FedEx's SOAP API.
http://python-fedex.readthedocs.org
BSD 3-Clause "New" or "Revised" License
156 stars 139 forks source link

FedEx RESTful API #170

Closed zesriver closed 1 year ago

zesriver commented 1 year ago

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 1 year ago

We would be very interested in this as well

MatinF commented 1 year ago

Has anybody found examples of 'full examples' of the REST API being used with Python e.g. to create international shipments?

MatinF commented 1 year ago

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 1 year ago

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.