244 lines
9.9 KiB
Python
244 lines
9.9 KiB
Python
#
|
|
# Python module implementing shipcloud.io REST API
|
|
# (C) 2021 by Harald Welte <laforge@gnumonks.org>
|
|
#
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
|
|
import sys
|
|
import logging
|
|
import requests
|
|
from requests.auth import HTTPBasicAuth
|
|
|
|
class ApiError(Exception):
|
|
"""Exception raised in case of a HTTP/REST API Error.
|
|
|
|
Attributes:
|
|
method -- HTTP method of the request causing the error
|
|
url -- URL of the HTTP request causing the error
|
|
sandbox -- request made in sandbox mode or not?
|
|
req_body -- json-serializable dict of request body causing the error
|
|
status -- HTTP status returned by REST API
|
|
errors -- list of string error messages returned by REST API
|
|
resp_body -- raw response body of failed rquest
|
|
"""
|
|
def __init__(self, method, url, sandbox, req_body, status, errors=[], resp_body=None):
|
|
self.method = method
|
|
self.url = url
|
|
self.sandbox = sandbox
|
|
self.req_body = req_body
|
|
self.status = status
|
|
self.errors = errors
|
|
self.resp_body = resp_body
|
|
def __str__(self):
|
|
sandbox_str = ' SANDBOX' if self.sandbox else ''
|
|
return "%s %s%s -> %s: %s" % (self.method, self.url, sandbox_str, self.status, self.errors)
|
|
|
|
class transport(object):
|
|
def __init__(self, api_key, api_key_sandbox=None, logger=logging.getLogger(__name__)):
|
|
self._api_key = api_key
|
|
self._auth = HTTPBasicAuth(self._api_key, '')
|
|
self._api_key_sandbox = api_key_sandbox or None
|
|
self._auth_sandbox = HTTPBasicAuth(self._api_key_sandbox, '') if self._api_key_sandbox else None
|
|
self._server_host = 'api.shipcloud.io'
|
|
self._server_port = 443
|
|
self._base_path= "/v1"
|
|
self._logger = logger
|
|
|
|
def _get_auth(self, sandbox=False):
|
|
if sandbox:
|
|
return self._auth_sandbox
|
|
else:
|
|
return self._auth
|
|
|
|
def _build_url(self, suffix):
|
|
return "https://%s:%u%s%s" % (self._server_host, self._server_port, self._base_path, suffix)
|
|
|
|
def rest_http(self, method, suffix, js = None, sandbox = False):
|
|
url = self._build_url(suffix)
|
|
sandbox_str = ' SANDBOX' if sandbox else ''
|
|
self._logger.debug("%s %s (%s)%s" % (method, url, str(js), sandbox_str))
|
|
resp = requests.request(method, url, json=js, auth=self._get_auth(sandbox))
|
|
self._logger.debug("-> %s - %s" % (resp, resp.text))
|
|
if not resp.ok:
|
|
self._logger.error("%s %s (%s)%s failed: %s - %s" % (method, url, str(js), sandbox_str, resp, resp.text))
|
|
errors = []
|
|
try:
|
|
resp_json = resp.json()
|
|
if 'errors' in resp_json:
|
|
errors = resp_json['errors']
|
|
except ValueError:
|
|
self._logger.error("response contains no valid json: %s" % (resp.text))
|
|
raise ApiError(method, url, sandbox, js, resp.status_code, errors, resp.text)
|
|
return resp.json()
|
|
|
|
def rest_post(self, suffix, js=None, sandbox=False):
|
|
return self.rest_http('POST', suffix, js, sandbox)
|
|
|
|
def rest_get(self, suffix, js=None, sandbox=False):
|
|
return self.rest_http('GET', suffix, js, sandbox)
|
|
|
|
def rest_delete(self, suffix, js=None, sandbox=False):
|
|
return self.rest_http('DELETE', suffix, js, sandbox)
|
|
|
|
class api(object):
|
|
def __init__(self, api_key, api_key_sandbox=None, logger=logging.getLogger(__name__)):
|
|
self._transport = transport(api_key, api_key_sandbox, logger)
|
|
|
|
def get_shipment_quote(self, shipment):
|
|
# quote request cannot be issued against sandbox, so always use production
|
|
res = self._transport.rest_post('/shipment_quotes', gen_quote_req(shipment))
|
|
return res
|
|
|
|
def create_shipment(self, shipment, gen_label=False):
|
|
# Assume if the user passed a sandbox API key, we use it for create shipment requests
|
|
sandbox = True if self._transport._auth_sandbox else False
|
|
sh = shipment.copy()
|
|
# don't send any shipping notifications to the real recipient when in sandbox
|
|
if sandbox:
|
|
sh['notification_mail'] = None
|
|
sh['create_shipping_label'] = gen_label
|
|
res = self._transport.rest_post('/shipments', sh, sandbox)
|
|
# {u'label_url': u'https://shipping-labels.shipcloud.io/shipments/a948e8c2/e3fb26be59/label/shipping_label_e3fb26be59.pdf', u'price': 0.0, u'id': u'e3fb26be59a68acd04d565dda027efd415ca8117', u'tracking_url': u'https://track.shipcloud.io/e3fb26be59a68acd04d565dda027efd415ca8117', u'carrier_tracking_no': u'1ZV306W00493609016'}
|
|
if 'label_url' in res:
|
|
r = requests.get(res['label_url'], stream=True)
|
|
if r.ok:
|
|
res['label_bin'] = r.content
|
|
if 'customs_declaration' in res and 'carrier_declaration_document_url' in res['customs_declaration']:
|
|
r = requests.get(res['customs_declaration']['carrier_declaration_document_url'], stream=True)
|
|
if r.ok:
|
|
res['customs_declaration_bin'] = r.content
|
|
return res
|
|
|
|
def delete_shipment(self, shipment_id):
|
|
sandbox = True if self._transport._auth_sandbox else False
|
|
res = self._transport.rest_delete('/shipments/%s' % shipment_id, None, sandbox)
|
|
return res
|
|
|
|
def create_pickup(self, pickup):
|
|
# Assume if the user passed a sandbox API key, we use it for create shipment requests
|
|
sandbox = True if self._transport._auth_sandbox else False
|
|
res = self._transport.rest_post('/pickup_requests', pickup, sandbox)
|
|
return res
|
|
|
|
|
|
def gen_customs_item(origin, desc, hts, qty, value, net_weight):
|
|
"""Generate a dict for a customs_declaration.item in accordance with
|
|
https://developers.shipcloud.io/reference/shipments_request_schema.html"""
|
|
customs_item = {
|
|
'origin_country': origin,
|
|
'description': desc,
|
|
'hs_tariff_number': str(hts),
|
|
'quantity': qty,
|
|
'value_amount': '%.2f' % value,
|
|
'net_weight': net_weight,
|
|
#'gross_weight': ,
|
|
}
|
|
return customs_item
|
|
|
|
def gen_customs_decl(currency, invoice_nr, net_total, items, importer_ref=None, exporter_ref=None):
|
|
"""Generate a dict for a customs_declaration in accordance with
|
|
https://developers.shipcloud.io/reference/shipments_request_schema.html"""
|
|
customs_decl = {
|
|
'contents_type': 'commercial_goods',
|
|
#'contents_explanation': ,
|
|
'currency' : currency,
|
|
'invoice_number': str(invoice_nr),
|
|
'total_value_amount': int(net_total),
|
|
'items': customs_items,
|
|
}
|
|
if importer_ref:
|
|
customs_decl['importer_reference'] = str(importer_ref)
|
|
if exporter_ref:
|
|
customs_decl['exporter_reference'] = str(exporter_ref)
|
|
return customs_decl
|
|
|
|
|
|
def gen_package(width_cm, length_cm, height_cm, weight_kgs, value=None, currency=None, descr=None):
|
|
"""Generate a dict for a package in accordance with
|
|
https://developers.shipcloud.io/reference/shipments_request_schema.html"""
|
|
package = {
|
|
'width': int(width_cm),
|
|
'length': int(length_cm),
|
|
'height': int(height_cm),
|
|
'weight': weight_kgs,
|
|
'type': 'parcel',
|
|
}
|
|
if value:
|
|
if currency == None:
|
|
currency = 'EUR'
|
|
package['declared_value'] = {
|
|
'amount': value,
|
|
'currency': currency,
|
|
}
|
|
if descr:
|
|
package['description'] = descr
|
|
return package
|
|
|
|
|
|
def gen_shipment(from_addr, to_addr, pkg, ref, carrier='ups', service='one_day', label_fmt='pdf_a5',
|
|
descr=None, customs_decl=None, incoterm='dap', notification_mail=None):
|
|
"""Generate a dict for a shipment in accordance with
|
|
https://developers.shipcloud.io/reference/shipments_request_schema.html"""
|
|
shipment = {
|
|
'from': from_addr,
|
|
'to': to_addr,
|
|
'carrier': carrier,
|
|
'service': service,
|
|
'package' : pkg,
|
|
'reference_number': ref,
|
|
'label': {
|
|
'format': label_fmt,
|
|
},
|
|
'notification_mail': notification_mail,
|
|
'incoterm': incoterm,
|
|
'create_shipping_label': False,
|
|
}
|
|
if descr:
|
|
shipment['description'] = descr
|
|
if customs_decl:
|
|
shipment['customs_declaration'] = customs_decl
|
|
return shipment
|
|
|
|
|
|
def _filter_dict(indict, permitted_keys):
|
|
"""Filter an input dictionary; keep only those keys listed in permitted_keys"""
|
|
outdict = {}
|
|
for k in permitted_keys:
|
|
if k in indict:
|
|
outdict[k] = indict[k]
|
|
return outdict
|
|
|
|
def gen_quote_req(shipment):
|
|
"""Generate a dict in accordance with
|
|
https://developers.shipcloud.io/reference/shipment_quotes_request_schema.html"""
|
|
# for some weird reason, the ShipmentQuoteRequest schema doesn't permit all
|
|
# the keys that are permitted when generating a label, making this more complicated
|
|
# than it should
|
|
permitted_sh_keys = [ 'carrier', 'service', 'to', 'from', 'package' ]
|
|
permitted_pkg_keys = [ 'width', 'height', 'length', 'weight', 'type' ]
|
|
permitted_addr_keys = [ 'street', 'street_no', 'city', 'zip_code', 'country' ]
|
|
|
|
# create a copy so we don't modify the input data
|
|
sh = shipment.copy()
|
|
sh['from'] = _filter_dict(sh['from'], permitted_addr_keys)
|
|
sh['to'] = _filter_dict(sh['to'], permitted_addr_keys)
|
|
sh['package'] = _filter_dict(sh['package'], permitted_pkg_keys)
|
|
|
|
return _filter_dict(sh, permitted_sh_keys)
|
|
|
|
def gen_pickup(addr, earliest, latest, shipment_ids, carrier='ups'):
|
|
"""Generate a dict for a pickup request in accordance with
|
|
https://developers.shipcloud.io/reference/pickup_requests_request_schema.html"""
|
|
permitted_addr_keys = [ 'company', 'first_name', 'last_name', 'care_of',
|
|
'street', 'street_no', 'city', 'zip_code', 'state', 'country', 'phone' ]
|
|
pickup = {
|
|
'carrier': carrier,
|
|
'pickup_time': {
|
|
'earliest': earliest,
|
|
'latest': latest,
|
|
},
|
|
'pickup_address': _filter_dict(addr, permitted_addr_keys)
|
|
}
|
|
return pickup
|