Browse Source

wpint: Address various pylint-reported issues

Harald Welte 5 months ago
  1. 49


@ -9,16 +9,19 @@
# 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
import hashlib
from datetime import datetime
import requests
from lxml import etree
from pytz import timezone
_logger = logging.getLogger(__name__)
class WarenpostInt(object):
"""Represents the Warenpost Internatoinal ReST interface."""
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
@ -55,6 +58,7 @@ class WarenpostInt(object):
# FIXME: merge with inema?
def gen_timestamp():
"""Generate a timestamp as used in the Warenpsost International API."""
de_zone = timezone("Europe/Berlin")
de_time =
return de_time.strftime("%d%m%Y-%H%M%S")
@ -82,21 +86,21 @@ class WarenpostInt(object):
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")
e_user_token = et.find(".//{}userToken")
e_wallet_balance = 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
self.user_token = e_user_token.text
self.wallet_balance = e_wallet_balance.text
_logger.debug("User Token: %s", self.user_token)"Wallet balance: %s", self.wallet_balance)
return e_user_token.text
def request(self, method, suffix, json=None, headers={}):
def request(self, method, suffix, json=None, headers=None):
"""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()
h = headers.copy() if headers else {}
_logger.debug("HTTP Request: %s %s: HDR: %s JSON: %s", method, url, h, json)
r = requests.request(method, url, json=json, headers=h)
@ -177,7 +181,8 @@ class WarenpostInt(object):
ret['recipientEmail'] =
return ret
def build_content_item(self, line_weight_g, line_value, qty, hs_code=None, origin_cc=None, desc=None):
def build_content_item(self, line_weight_g, line_value, qty, hs_code=None, origin_cc=None,
"""Build an 'content item' in the language of the WaPoInt API. Represents one
line on the customs form."""
ret = {
@ -193,8 +198,8 @@ class WarenpostInt(object):
ret['contentPieceOrigin'] = origin_cc
return ret
def build_item(self, product, sender, receiver, weight_grams, amount=0, currency='EUR',
def build_item(self, product, sender, recipient, weight_grams, amount=0, currency='EUR',
"""Build an 'item' in the language of the WaPoInt API. Represents one shipment."""
ret = {
'product': str(product),
@ -207,26 +212,26 @@ class WarenpostInt(object):
# merge in the sender and recipient fields
if len(contents):
if contents:
ret['contents'] = contents
return ret
def build_order(self, items, contactName, orderStatus='FINALIZE'):
def build_order(self, items, contact_name, order_status='FINALIZE'):
"""Build an 'order' in the language of the WaPoInt API. Consists of multiple shipments."""
ret = {
'customerEkp': self.ekp,
'orderStatus': orderStatus,
'orderStatus': order_status,
'paperwork': {
'contactName': contactName,
'contactName': contact_name,
'awbCopyCount': 1,
'items': items
return ret
def api_create_order(self, items, contactName, orderStatus='FINALIZE'):
def api_create_order(self, items, contact_name, order_status='FINALIZE'):
"""Issue an API request to create an order consisting of items."""
order = self.build_order(items, contactName=contactName, orderStatus=orderStatus)
order = self.build_order(items, contact_name=contact_name, order_status=order_status)"Order: %s", order)
r = self.request('POST', 'orders', json = order)
# TODO: figure out the AWB and the (item, barcode, voucherId, ...) for the items