import requests
from TruckstopLibrary.Load import Load, Pagination, SortCriteria, SearchCriteria
import json
baseURLStage = "https://api-int.truckstop.com"
baseURLProd = "https://api.truckstop.com"
def PostLoad(load: Load, token: str, isSandbox):
if isSandbox:
url = baseURLStage + "/loadmanagement/v2/load"
else:
url = baseURLProd + "/loadmanagement/v2/load"
headers = {
"accept": "application/json",
"Authorization": "bearer " + token,
"content-type": "application/json"
}
payload = json.dumps(load.to_dict())
response = requests.post(url,data=payload, headers=headers)
if (response.status_code == 201):
return ParseResponseToLoad(response.text), None
else:
return None, response.text
def UpdateLoad(load: Load, token: str, isSandbox):
if isSandbox:
url = baseURLStage + "/loadmanagement/v2/load/" + load.loadId
else:
url = baseURLProd + "/loadmanagement/v2/load/" + load.loadId
headers = {
"accept": "application/json",
"Authorization": "bearer " + token,
"content-type": "application/json"
}
payload = json.dumps(load.to_dict())
response = requests.put(url,data=payload, headers=headers)
if (response.status_code == 200):
return ParseResponseToLoad(response.text), None
else:
return None, response.text
def DeleteLoad(load: Load, token: str, isSandbox):
if isSandbox:
url = baseURLStage + "/loadmanagement/v2/load/" + load.loadId
else:
url = baseURLProd + "/loadmanagement/v2/load/" + load.loadId
headers = {
"accept": "application/json",
"Authorization": "bearer " + token,
"content-type": "application/json"
}
payload = {
"reason": "2",
"comment": "Not Needed"
}
response = requests.delete(url,json=payload, headers=headers)
if(response.status_code==204):
return True
else:
print(response.text)
return False
def RefreshLoad(load: Load, token: str, isSandbox):
if isSandbox:
url = baseURLStage + "/loadmanagement/v2/load/refresh/integrator"
else:
url = baseURLProd + "/loadmanagement/v2/load/refresh/integrator"
headers = {
"Authorization": "bearer " + token,
"accept": "application/json",
"content-type": "application/json"
}
loads = [f'{load.loadId}']
response = requests.post(url, json=loads, headers=headers)
print(response.text)
def SearchLoads(pagination: Pagination, sortCriteria: SortCriteria, searchCriteria: SearchCriteria, token: str, isSandbox=True):
if isSandbox:
url = baseURLStage + "/loadmanagement/v2/load/search"
else:
url = baseURLProd + "/loadmanagement/v2/load/search"
headers = {
"Authorization": "bearer " + token,
"accept": "application/json",
"content-type": "application/json"
}
payload = {
"pagination": pagination.to_dict(),
"searchCriteria": [searchCriteria.to_dict()],
"sortCriteria": [sortCriteria.to_dict()]
}
response = requests.post(url, json=payload, headers=headers)
return CreateLoadListFromSearchResponse(response.text)
def GetLoadDetails(load: Load, token: str, isSandbox):
if isSandbox:
url = baseURLStage + "/loadmanagement/v2/load/" + load.loadId
else:
url = baseURLProd + "/loadmanagement/v2/load/" + load.loadId
headers = {
"accept": "application/json",
"Authorization": "bearer " + token,
"content-type": "application/json"
}
response = requests.get(url, headers=headers)
if(response.status_code==200):
return ParseResponseToLoad(response.text)
else:
print(response.text)
return False
def CreateLoadListFromSearchResponse(response):
data = json.loads(response)
loads = []
try:
# Iterate over each load in the response
for load_info in data['data']:
loads.append(ParseResponseToLoad(load_info))
except Exception as e:
print("Exception: " + str(e))
return loads
def ParseResponseToLoad(jsonString):
if not isinstance(jsonString, dict):
response_data = json.loads(jsonString)
else:
response_data = jsonString
termsAndConditionsData = response_data.get('termsAndConditions')
termsAndConditions = None
if termsAndConditionsData:
termsAndConditions = Load.TermsAndConditions(
id = termsAndConditionsData.get('id'),
name = termsAndConditionsData.get('name')
)
loadActionAttributesData = response_data.get('loadActionAttributes')
loadActionAttributes = None
if loadActionAttributesData:
loadActionAttributes = Load.LoadActionAttributes(
loadActionId=loadActionAttributesData.get('loadActionId'),
loadActionOption=loadActionAttributesData.get('loadActionOption')
)
dimensionalData = response_data.get('dimensional')
dimensional = None
if dimensionalData:
dimensional = Load.Dimensional(
length = dimensionalData.get('length') if (dimensionalData.get('length') is not None and float(dimensionalData.get('length')) > 0) else None,
width = dimensionalData.get('width') if (dimensionalData.get('width') is not None and float(dimensionalData.get('width')) > 0) else None,
height = dimensionalData.get('height') if (dimensionalData.get('height') is not None and float(dimensionalData.get('height')) > 0) else None,
weight = dimensionalData.get('weight') if (dimensionalData.get('weight') is not None and float(dimensionalData.get('weight')) > 0) else None,
palletCount = dimensionalData.get('palletCount') if (dimensionalData.get('palletCount') is not None and float(dimensionalData.get('palletCount')) > 0) else None,
pieceCount = dimensionalData.get('pieceCount') if (dimensionalData.get('pieceCount') is not None and float(dimensionalData.get('pieceCount')) > 0) else None,
cube = dimensionalData.get('cube') if (dimensionalData.get('cube') is not None and float(dimensionalData.get('cube')) > 0) else None
)
# Parse equipmentAttributes
equipment_attributes_data = response_data.get('equipmentAttributes')
equipment_attributes = None
if equipment_attributes_data:
equipment_attributes = Load.EquipmentAttributes(
equipmentTypeId=equipment_attributes_data.get('equipmentTypeId'),
equipmentOptions=equipment_attributes_data.get('equipmentOptions'),
transportationModeId=equipment_attributes_data.get('transportationModeId'),
otherEquipmentNeeds=equipment_attributes_data.get('otherEquipmentNeeds')
)
rate_attributes_data = response_data.get('rateAttributes')
rate_attributes = None
if rate_attributes_data:
rate_attributes = Load.RateAttributes(
postedAllInRate=Load.Currency(rate_attributes_data.get('postedAllInRate').get('amount'),"USD"),
tenderAllInRate=Load.Currency(rate_attributes_data.get('tenderAllInRate').get('amount'),"USD")
)
# Parse loadStops
load_stops_data = response_data.get('loadStops', [])
load_stops = []
for stop_data in load_stops_data:
location_data = stop_data.get('location')
location = Load.Location(
id=location_data.get('id'),
locationName=location_data.get('locationName'),
city=location_data.get('city'),
state=location_data.get('state'),
streetAddress1=location_data.get('streetAddress1'),
streetAddress2=location_data.get('streetAddress2'),
countryCode=location_data.get('countryCode'),
postalCode=location_data.get('postalCode'),
latitude=location_data.get('latitude'),
longitude=location_data.get('longitude')
)
load_stop = Load.LoadStop(
id=stop_data.get('id'),
type=stop_data.get('type'),
sequence=stop_data.get('sequence'),
earlyDateTime=stop_data.get('earlyDateTime'),
lateDateTime=stop_data.get('lateDateTime'),
location=location,
contactName=stop_data.get('contactName'),
contactPhone=stop_data.get('contactPhone'),
stopNotes=stop_data.get('stopNotes'),
referenceNumbers=stop_data.get('referenceNumbers')
)
load_stops.append(load_stop)
# Construct the Load object
load = Load(
loadId=response_data.get('loadId'),
legacyLoadId=response_data.get('legacyLoadId'),
postAsUserId=response_data.get('postAsUserId'),
postAsUser=response_data.get('postAsUser'),
createdBy=response_data.get('createdBy'),
updatedBy=response_data.get('updatedBy'),
createDateTime=response_data.get('createDateTime'),
updateDateTime=response_data.get('updateDateTime'),
loadState=response_data.get('loadState'),
loadStateReason=response_data.get('loadStateReason'),
tenderId=response_data.get('tenderId'),
carrierName=response_data.get('carrierName'),
loadActivity=response_data.get('loadActivity', []),
source=response_data.get('source'),
loadSource=response_data.get('loadSource'),
integratorLoadUrl=response_data.get('integratorLoadUrl'),
nextRefreshAvailableAt=response_data.get('nextRefreshAvailableAt'),
equipmentAttributes=equipment_attributes,
commodityId=response_data.get('commodityId'),
loadStops=load_stops, # The list of LoadStop objects
note=response_data.get('note'),
freightClassId=response_data.get('freightClassId'),
loadNumber=response_data.get('loadNumber'),
loadTrackingRequired=response_data.get('loadTrackingRequired'),
rateAttributes=rate_attributes,
customData=response_data.get('customData', []),
distance=response_data.get('distance'),
dimensional=dimensional,
loadActionAttributes=loadActionAttributes,
loadLabel=response_data.get('loadLabel'),
tenderNotes=response_data.get('tenderNotes'),
loadReferenceNumbers=response_data.get('loadReferenceNumbers', []),
termsAndConditions=termsAndConditions,
carrierTenderGroupIds=response_data.get('carrierTenderGroupIds', []),
isPrivate=response_data.get('isPrivate'),
)
return loadAll
Pages