Browse Source

Add 'wpint' module for Warenpost International ReST-API

This is a new module conforming to revision 1.03 of the
above-mentioned ReST-API.
Harald Welte 5 months ago
  1. 1
  2. 247


@ -2,3 +2,4 @@
from .inema import Internetmarke
from .inema import ProductInformation
from .inema import __version__
from .wpint import WarenpostInt


@ -0,0 +1,247 @@
# -*- coding: utf-8 -*-
# Implementation of the "Warenpost International ReST-API"
# This is a completely different interface that Deutsche Post came up many years
# after the "1C4A" Internetmarke API. For some strange reason they didn't
# extend the old Internetmarke API to add support for the harmonized label and
# the electronic customs declaration. Instead, they decided to implement a
# completely different API with different standards (REST vs. SOAP) with
# literally nothing in common to the old Internetmarke API
from datetime import datetime, date
from pytz import timezone
import requests
from lxml import etree
import json
import logging
_logger = logging.getLogger(__name__)
class WarenpostInt(object):
def __init__(self, partner_id, key, ekp, pk_email, pk_passwd, key_phase="1", sandbox = False):
self.sandbox = sandbox
self.partner_id = 'DP_LT' if sandbox else partner_id
self.key = key
self.ekp = ekp
self.key_phase = '1' if sandbox else key_phase
self.pk_email = pk_email
self.pk_passwd = pk_passwd
if sandbox:
self.auth_url = ''
self.url = ''
self.auth_url = ''
self.url = ''
self.user_token = None
self.wallet_balance = None
# FIXME: merge with inema?
def compute_1c4a_hash(partner_id, req_ts, key_phase, key):
""" Compute 1C4A request hash accordig to Section 4 of service description. """
# trim leading and trailing spaces of each argument
partner_id = partner_id.strip()
req_ts = req_ts.strip()
key_phase = key_phase.strip()
key = key.strip()
# concatenate with "::" separator
inp = "%s::%s::%s::%s" % (partner_id, req_ts, key_phase, key)
# compute MD5 hash as 32 hex nibbles
md5_hex = hashlib.md5(inp.encode('utf8')).hexdigest()
# return the first 8 characters
return md5_hex[:8]
# FIXME: merge with inema?
def gen_timestamp():
de_zone = timezone("Europe/Berlin")
de_time =
return de_time.strftime("%d%m%Y-%H%M%S")
def gen_headers(self):
"""Generate the HTTP headers required for the API."""
ret = {
'KEY_PHASE': self.key_phase,
'PARTNER_ID': self.partner_id,
'Authorization': 'Bearer %s' % self.user_token
if self.sandbox:
ret['REQUEST_TIMESTAMP'] = '16082018-122210'
ret['PARTNER_SIGNATURE'] = '9d7c35be'
timestamp = self.gen_timestamp()
sig = self.compute_1c4a_hash(self.partner_id, timestamp, self.key_phase, self.key)
ret['REQUEST_TIMESTAMP'] = timestamp
return ret
def get_token(self):
"""Get an Access Token for further API requests."""
url = "%s/%s" % (self.auth_url, 'auth/accesstoken')
auth = requests.auth.HTTPBasicAuth(self.pk_email, self.pk_passwd)
ret = requests.request('GET', url, headers=self.gen_headers(), auth=auth)
et = etree.XML(ret.content)
e_userToken = et.find(".//{}userToken")
e_walletBalance = et.find(".//{}walletBalance")
# update status + return token
self.user_token = e_userToken.text
self.wallet_balance = e_walletBalance.text
_logger.debug("User Token: %s" % (self.user_token))"Wallet balance: %s" % (self.wallet_balance))
return e_userToken.text
def request(self, method, suffix, json=None, headers={}):
"""Wrapper for issuing HTTP requests against the API.
This internally generates all required headers, including Authorization."""
url = "%s/%s" % (self.url, suffix)
# FIXME: automatically ensure we have a [current] user_token
h = headers.copy()
_logger.debug("HTTP Request: %s %s: HDR: %s JSON: %s", method, url, h, json)
r = requests.request(method, url, json=json, headers=h)
_logger.debug("HTTP Response: %s", r.content)
return r
class Address(object):
"""Common Representation of a postal address. In their infinite cluelessness,
the developes of the Warenpost International API decided it's a good idea to
use a flat, non-hierarchical structure with different names of fields for
sender and recipient."""
def __init__(self, name, addr_lines, city, country_code, postal_code='', state=None,
phone=None, fax=None, email=None):
if len(addr_lines) > 3:
raise ValueError('Maximum number of 3 Address Lines supported')
if len(name) > 30:
raise ValueError('Maximum length of name is 30 chars')
if phone and len(phone) > 15:
raise ValueError('Maximum length of phone number is 15 chars')
if fax and len(fax) > 15:
raise ValueError('Maximum length of fax number is 15 chars')
if email and len(email) > 50:
raise ValueError('Maximum length of email address is 50 chars')
for l in addr_lines:
if len(l) > 40:
raise ValueError('Maximum length of address lines is 40 chars') = name
self.addr_lines = addr_lines = city
self.postal_code = postal_code
self.country_code = country_code
self.state = state = phone
self.fax = fax = email
def as_sender(self):
"""Represent an Address object as JSON fields of a sender."""
if len(self.addr_lines) > 2:
raise ValueError('Maximum number of 2 Address Lines supported')
for l in self.addr_lines:
if len(l) > 30:
raise ValueError('Maximum length of address lines is 30 chars')
ret = {
'senderAddressLine1': self.addr_lines[0],
'senderAddressLine2': self.addr_lines[1] if len(self.addr_lines) > 1 else '',
'senderPostalCode': self.postal_code,
'senderCountry': self.country_code,
ret['senderPhone'] =
ret['senderEmail'] =
return ret
def as_recipient(self):
"""Represent an Address object as JSON fields of a sender."""
ret = {
'addressLine1': self.addr_lines[0],
'postalCode': self.postal_code,
'destinationCountry': self.country_code,
if len(self.addr_lines) > 1:
ret['addressLine2'] = self.addr_lines[1]
if len(self.addr_lines) > 2:
ret['addressLine3'] = self.addr_lines[2]
if self.state:
ret['state'] = self.state
ret['recipientPhone'] =
if self.fax:
ret['recipientFax'] = self.fax
ret['recipientEmail'] =
return ret
def build_content_item(self, line_weight_g, line_value, qty, hs_code=None, origin_cc=None, desc=None):
"""Build an 'content item' in the language of the WaPoInt API. Represents one
line on the customs form."""
ret = {
'contentPieceNetweight': line_weight_g,
'contentPieceValue': str(int(line_value)),
'contentPieceAmount': int(qty),
if hs_code:
ret['contentPieceHsCode'] = str(hs_code)
if desc:
ret['contentPieceDescription'] = str(desc)
if origin_cc:
ret['contentPieceOrigin'] = origin_cc
return ret
def build_item(self, product, sender, receiver, weight_grams, amount=0, currency='EUR',
"""Build an 'item' in the language of the WaPoInt API. Represents one shipment."""
ret = {
'product': str(product),
'serviceLevel': 'STANDARD',
'shipmentAmount': int(amount),
'shipmentCurrency': currency,
'shipmentGrossWeight': int(weight_grams),
'shipmentNaturetype': 'SALE_GOODS',
# merge in the sender and recipient fields
if len(contents):
ret['contents'] = contents
return ret
def build_order(self, items, contactName, orderStatus='FINALIZE'):
"""Build an 'order' in the language of the WaPoInt API. Consists of multiple shipments."""
ret = {
'customerEkp': self.ekp,
'orderStatus': orderStatus,
'paperwork': {
'contactName': contactName,
'awbCopyCount': 1,
'items': items
return ret
def api_create_order(self, items, contactName, orderStatus='FINALIZE'):
"""Issue an API request to create an order consisting of items."""
order = self.build_order(items, contactName=contactName, orderStatus=orderStatus)"Order: %s", order)
r = self.request('POST', 'orders', json = order)
# TODO: figure out the AWB and the (item, barcode, voucherId, ...) for the items
json_resp = r.json()
#print(json.dumps(json_resp, indent=4))"Response: %s", json_resp)
# TODO: download the PDF for each AWB
return json_resp
def api_get_item_label(self, item_id, accept='application/pdf'):
"""Download the label for a given item. Returns PDF as 'bytes'"""
r = self.request('GET', 'items/%s/label' % item_id, headers={'Accept': accept})
return r.content
def api_get_item_labels(self, awb, accept='application/pdf'):
"""Download the labels for all items in a given AWB. Returns PDF as 'bytes'"""
r = self.request('GET', 'shipments/%s/itemlabels' % awb, headers={'Accept': accept})
return r.content