REST API - unable to execute Upload Documents

[MatinF]

I am trying to finalize a migration of the full 'international shipment' example from the SOAP API to the REST API. Aiming to ideally share this on github afterwards as I've been unable to find any full Python examples.

To finalize this, I am trying to get the Upload Documents API integrated, but struggling - hope some here can help.

FedEx Docs on Upload Documents

From the Docs, the details are as below:

Here, the suggested payload is as follows:

document: {"workflowName":"ETDPreshipment","name":"file.txt","contentType":"text/plain","meta":{"shipDocumentType":"COMMERCIAL_INVOICE","originCountryCode":"US","destinationCountryCode":"IN"}}
attachment: file.txt

My attempted code

I've included two functions relevant below for this:

The first is my function for creating an oauth2 authentication token, which works as intended for my Create Shipment API call.

# create access token for use in shipment creation and document upload
def get_access_token(self):
import json, requests

url = "https://apis.fedex.com/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

The second is my attempted code for the Upload Documents functionality:

# upload a single commercial invoice as ETD
def doc_upload_test(self):
    import requests
    from requests_toolbelt.multipart.encoder import MultipartEncoder

    url = "https://documentapi.prod.fedex.com/documents/v1/etds/upload"

    filename = "file.pdf"
    file_contents = open(filename, 'rb').read()

    payload = {
        'document': ('document.json', '{"workflowName":"ETDPreshipment","name":"file.pdf","contentType":"application/pdf","meta":{"shipDocumentType":"COMMERCIAL_INVOICE","originCountryCode":"US","destinationCountryCode":"IN"}}', 'application/json'),
        'attachment': (filename, file_contents, 'application/pdf')
    }

    m = MultipartEncoder(fields=payload)

    headers = {
        'x-customer-transaction-id': "",
        'Content-Type': "multipart/form-data",
        'Authorization': f"Bearer {self.access_token}"
    }

    response = requests.post(url, data=m, headers=headers)

    print(response.text)

The errors

The above code does not work and results in the below error:

<?xml version="1.0" encoding="UTF-8"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
    <soapenv:Body>
        <soapenv:Fault>
            <faultcode>soapenv:Server</faultcode>
            <faultstring>Error in assertion processing</faultstring>
            <faultactor>ssg</faultactor>
            <detail>
                <l7:policyResult
                    status="Multipart message preamble too large" xmlns:l7="[http://www.layer7tech.com/ws/policy/fault"/](http://www.layer7tech.com/ws/policy/fault%22/)>
            </detail>
        </soapenv:Fault>
    </soapenv:Body>
</soapenv:Envelope>

Simpler attempt with TXT file

I also tried a simpler attempt with below request:

response = requests.post(
            url,
            files=dict(
                document='{"workflowName":"ETDPreshipment","name":"file.txt","contentType":"text/plain","meta":{"shipDocumentType":"COMMERCIAL_INVOICE","originCountryCode":"DK","destinationCountryCode":"BE"}',
                attachment=("file.txt", "this is a test", "text/plain"),
            ),
            headers=headers,
        )

This results in a different error as below:

{  
 "customerTransactionId": "",
   "errors":[  
         {  
            "code":"AUTHENTICATION.TOKEN.INVALID",
            "message":"Invalid token passed in the request"
         }
      ]
}

I have tested this with both draft and production credentials to no avail. Any inputs are welcome!

[MatinF]

I finally managed to solve this. Below is the full FedEx class I created for the REST API:

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()
        file_content_b64 = binascii.b2a_base64(file_content)
        file_content_b64.decode("cp1250")

        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_b64, "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 ""