Added default private key to constructor for ease of use + added extra headers to invoicing

Author: koen-serry

README.md

@@ -51,6 +51,7 @@ and configure your API key which you can find in the [Twikey merchant interface]
 ```python
 import twikey
 
+APIKEY = 'apikey_as_found_in_twikey'
 twikeyClient = twikey.TwikeyClient(APIKEY, "https://api.beta.twikey.com")
 ``` 
 
@@ -60,7 +61,11 @@ Invite a customer to sign a SEPA mandate using a specific behaviour template (ct
 the behaviour or flow that the customer will experience. This 'ct' can be found in the template section of the settings.
 
 ```python
-invite = twikeyClient.document.create({
+import twikey
+
+ct = 123 # See settings/profile in twikey
+
+invite = twikey.TwikeyClient.document.create({
    "ct": ct,
    "email": "info@twikey.com",
    "firstname": "Info",
@@ -81,24 +86,26 @@ think of as reading out a queue. Since it'll return you the changes since the la
 import twikey
 
 class MyDocumentFeed(twikey.DocumentFeed):
-    def newDocument(self, doc):
+    def newDocument(self, doc, evt_time):
         print("new ", doc["MndtId"])
 
-    def updatedDocument(self, orinalDocNumber, doc, reason):
+    def updatedDocument(self, original_mandate_number, doc, reason, evt_time):
         print("update ", doc["MndtId"], "b/c", reason["Rsn"])
 
-    def cancelDocument(self, docNumber, reason):
-        print("cancelled ", docNumber, "b/c", reason["Rsn"])
+    def cancelDocument(self, doc_number, reason, evt_time):
+        print("cancelled ", doc_number, "b/c", reason["Rsn"])
 
-twikeyClient.document.feed(MyDocumentFeed())
+twikey.TwikeyClient.document.feed(MyDocumentFeed())
 ```
 
 ## Transactions
 
 Send new transactions and act upon feedback from the bank.
 
 ```python
-tx = twikeyClient.transaction.create({
+import twikey
+
+tx = twikey.TwikeyClient.transaction.create({
    "mndtId" : "CORERECURRENTNL16318",
    "message" : "Test Message",
    "ref" : "Merchant Reference",
@@ -116,7 +123,7 @@ class MyFeed(twikey.TransactionFeed):
     def transaction(self, transaction):
         print("TX ", transaction.ref, transaction.state)
 
-twikeyClient.transaction.feed(MyFeed())
+twikey.TwikeyClient.transaction.feed(MyFeed())
 ```
 
 ## Webhook ##
@@ -125,13 +132,15 @@ When wants to inform you about new updates about documents or payments a `webhoo
 
 ```python
 import Flask 
-import twikey 
+import urllib
+import twikey
 
+APIKEY = 'apikey_as_found_in_twikey'
 app = Flask(__name__)
 
 @app.route('/webhook', methods=['GET'])
 def webhook(request):
-   payload = unquote(request.query_string)
+   payload = urllib.parse.unquote(request.query_string)
    received_sign = request.headers.get('X-Signature')
    if not received_sign:
       return False

requirements.txt

@@ -0,0 +1,2 @@
+urllib3~=1.26.5
+requests~=2.25.1

setup.py

@@ -10,7 +10,7 @@ def get_long_description():
 
 setup(
     name="twikey-api-python",
-    version="v0.1.4",
+    version="v0.1.7",
     description="Python interface with the Twikey api",
     author="Twikey",
     author_email="support@twikey.com",

tests/test_document.py

@@ -43,16 +43,16 @@ def test_feed(self):
 
 
 class MyDocumentFeed(twikey.DocumentFeed):
-    def newDocument(self, doc, evtTime):
-        print("Document created   ", doc["MndtId"], "@", evtTime)
+    def newDocument(self, doc, evt_time):
+        print("Document created   ", doc["MndtId"], "@", evt_time)
 
-    def updatedDocument(self, original_number, doc, reason, evtTime):
+    def updatedDocument(self, original_number, doc, reason, evt_time):
         print(
-            "Document updated   ", original_number, "b/c", reason["Rsn"], "@", evtTime
+            "Document updated   ", original_number, "b/c", reason["Rsn"], "@", evt_time
         )
 
-    def cancelDocument(self, number, reason, evtTime):
-        print("Document cancelled ", number, "b/c", reason["Rsn"], "@", evtTime)
+    def cancelDocument(self, number, reason, evt_time):
+        print("Document cancelled ", number, "b/c", reason["Rsn"], "@", evt_time)
 
 
 if __name__ == "__main__":

tox.ini

@@ -26,6 +26,7 @@ deps =
     pytest-cov >= 2.8.1, < 2.11.0
     pytest-mock >= 2.0.0
     pytest-xdist >= 1.31.0
+    requests ~= 2.25.1
 commands = pytest --cov {posargs:-n auto}
 # compilation flags can be useful when prebuilt wheels cannot be used, e.g.
 # PyPy 2 needs to compile the `cryptography` module. On macOS this can be done
@@ -37,6 +38,6 @@ passenv = LDFLAGS CFLAGS
 [testenv:fmt]
 description = run code formatting using black
 basepython = python3.9
-deps = black==20.8b1
+deps = black==22.6.0
 commands = black . {posargs}
 skip_install = true

twikey/client.py

@@ -1,16 +1,11 @@
-import binascii
-import hmac
-import struct
-import time
-import datetime
-
 import requests
 
 from .document import Document
 from .transaction import Transaction
 from .paylink import Paylink
 from .invoice import Invoice
 import logging
+import datetime
 
 
 class TwikeyClient(object):
@@ -19,7 +14,7 @@ class TwikeyClient(object):
     api_token = None  # Once authenticated
     merchant_id = 0  # Once authenticated
     private_key = None
-    vendorPrefix = "own"
+    vendorPrefix = b"own"
     api_base = "https://api.twikey.com"
 
     document = None
@@ -32,9 +27,11 @@ def __init__(
         api_key,
         base_url="https://api.twikey.com",
         user_agent="twikey-python/v0.1.0",
+        private_key=None,
     ) -> None:
         self.user_agent = user_agent
         self.api_key = api_key
+        self.private_key = private_key
         self.api_base = base_url
         self.merchant_id = 0
         self.document = Document(self)
@@ -48,14 +45,17 @@ def instance_url(self, url=""):
 
     def get_totp(self, vendorPrefix, secret):
         """Return the Time-Based One-Time Password for the current time, and the provided secret (base32 encoded)"""
-        secret = bytearray(vendorPrefix) + binascii.unhexlify(secret)
-        counter = struct.pack(">Q", int(time.time()) // 30)
-
         import hashlib
+        import struct
+        import binascii
+        import hmac
+        import time
 
-        hash = hmac.new(secret, counter, hashlib.sha256).digest()
-        offset = ord(hash[19]) & 0xF
+        secret = vendorPrefix + binascii.unhexlify(secret)
+        counter = struct.pack(">Q", int(time.time()) // 30)
 
+        hash = hmac.new(secret, counter, hashlib.sha256).digest()
+        offset = hash[19] & 0xF
         return (
             struct.unpack(">I", hash[offset : offset + 4])[0] & 0x7FFFFFFF
         ) % 100000000

twikey/document.py

@@ -13,12 +13,12 @@ def create(self, data):
         data = data or {}
         self.client.refreshTokenIfRequired()
         response = requests.post(url=url, data=data, headers=self.client.headers())
-        jsonResponse = response.json()
+        json_response = response.json()
         if "ApiErrorCode" in response.headers:
-            error = jsonResponse
+            error = json_response
             raise Exception("Error invite : %s" % error)
-        self.logger.debug("Added new mandate : %s" % jsonResponse["mndtId"])
-        return jsonResponse
+        self.logger.debug("Added new mandate : %s" % json_response["mndtId"])
+        return json_response
 
     def update(self, data):
         url = self.client.instance_url("/mandate/update")
@@ -45,7 +45,7 @@ def cancel(self, mandate_number, reason):
             error = response.json()
             raise Exception("Error invite : %s" % error)
 
-    def feed(self, documentFeed):
+    def feed(self, document_feed):
         url = self.client.instance_url("/mandate")
 
         self.client.refreshTokenIfRequired()
@@ -54,39 +54,39 @@ def feed(self, documentFeed):
         if "ApiErrorCode" in response.headers:
             error = response.json()
             raise Exception("Error feed : %s" % error)
-        feedResponse = response.json()
-        while len(feedResponse["Messages"]) > 0:
-            self.logger.debug("Feed handling : %d" % (len(feedResponse["Messages"])))
-            for msg in feedResponse["Messages"]:
+        feed_response = response.json()
+        while len(feed_response["Messages"]) > 0:
+            self.logger.debug("Feed handling : %d" % (len(feed_response["Messages"])))
+            for msg in feed_response["Messages"]:
                 if "AmdmntRsn" in msg:
                     mndt_id_ = msg["OrgnlMndtId"]
-                    self.logger.debug("Feed update : %s" % (mndt_id_))
+                    self.logger.debug("Feed update : %s" % mndt_id_)
                     mndt_ = msg["Mndt"]
                     rsn_ = msg["AmdmntRsn"]
-                    documentFeed.updatedDocument(mndt_id_, mndt_, rsn_, msg["EvtTime"])
+                    document_feed.updatedDocument(mndt_id_, mndt_, rsn_, msg["EvtTime"])
                 elif "CxlRsn" in msg:
                     self.logger.debug("Feed cancel : %s" % (msg["OrgnlMndtId"]))
-                    documentFeed.cancelDocument(
+                    document_feed.cancelDocument(
                         msg["OrgnlMndtId"], msg["CxlRsn"], msg["EvtTime"]
                     )
                 else:
                     self.logger.debug("Feed create : %s" % (msg["Mndt"]))
-                    documentFeed.newDocument(msg["Mndt"], msg["EvtTime"])
+                    document_feed.newDocument(msg["Mndt"], msg["EvtTime"])
             response = requests.get(url=url, headers=self.client.headers())
             if "ApiErrorCode" in response.headers:
                 error = response.json()
                 raise Exception(
                     "Error invite : %s - %s" % (error["code"], error["message"])
                 )
-            feedResponse = response.json()
+            feed_response = response.json()
 
 
 class DocumentFeed:
-    def newDocument(self, doc, evtTime):
+    def newDocument(self, doc, evt_time):
         pass
 
-    def updatedDocument(self, original_mandate_number, doc, reason, evtTime):
+    def updatedDocument(self, original_mandate_number, doc, reason, evt_time):
         pass
 
-    def cancelDocument(self, docNumber, reason, evtTime):
+    def cancelDocument(self, doc_number, reason, evt_time):
         pass

twikey/invoice.py

@@ -8,20 +8,26 @@ def __init__(self, client) -> None:
         self.client = client
         self.logger = logging.getLogger(__name__)
 
-    def create(self, data):
+    def create(self, data, origin=False, purpose=False, manual=False):
         url = self.client.instance_url("/invoice")
         data = data or {}
         self.client.refreshTokenIfRequired()
         headers = self.client.headers("application/json")
+        if origin:
+            headers["X-PARTNER"] = origin
+        if purpose:
+            headers["X-Purpose"] = purpose
+        if manual:
+            headers["X-MANUAL"] = "true"
         response = requests.post(url=url, json=data, headers=headers)
-        jsonResponse = response.json()
+        json_response = response.json()
         if "ApiErrorCode" in response.headers:
-            error = jsonResponse
+            error = json_response
             raise Exception("Error creating : %s" % error)
-        self.logger.debug("Added invoice : %s" % jsonResponse["url"])
-        return jsonResponse
+        self.logger.debug("Added invoice : %s" % json_response["url"])
+        return json_response
 
-    def feed(self, invoiceFeed, *includes):
+    def feed(self, invoice_feed, *includes):
         _includes = ""
         for include in includes:
             _includes += "&include=" + include
@@ -36,17 +42,17 @@ def feed(self, invoiceFeed, *includes):
                 "Error feed : %s - %s"
                 % (response.headers["ApiErrorCode"], response.headers["ApiError"])
             )
-        feedResponse = response.json()
-        while len(feedResponse["Invoices"]) > 0:
-            self.logger.debug("Feed handling : %d" % (len(feedResponse["Invoices"])))
-            for invoice in feedResponse["Invoices"]:
+        feed_response = response.json()
+        while len(feed_response["Invoices"]) > 0:
+            self.logger.debug("Feed handling : %d" % (len(feed_response["Invoices"])))
+            for invoice in feed_response["Invoices"]:
                 self.logger.debug("Feed handling : %s" % invoice)
-                invoiceFeed.invoice(invoice)
+                invoice_feed.invoice(invoice)
             response = requests.get(url=url, headers=self.client.headers())
             if "ApiErrorCode" in response.headers:
                 error = response.json()
                 raise Exception("Error feed : %s" % error)
-            feedResponse = response.json()
+            feed_response = response.json()
 
     def geturl(self, invoice_id):
         return "%s/%s/%s" % (

twikey/paylink.py

@@ -16,7 +16,7 @@ def create(self, data):
             raise Exception("Error creating paylink : %s" % error)
         return response.json()
 
-    def feed(self, paylinkFeed):
+    def feed(self, paylink_feed):
         url = self.client.instance_url("/payment/link/feed")
 
         self.client.refreshTokenIfRequired()
@@ -25,15 +25,15 @@ def feed(self, paylinkFeed):
         if "ApiErrorCode" in response.headers:
             error = response.json()
             raise Exception("Error feed : %s" % error)
-        feedResponse = response.json()
-        while len(feedResponse["Links"]) > 0:
-            for msg in feedResponse["Links"]:
-                paylinkFeed.paylink(msg)
+        feed_response = response.json()
+        while len(feed_response["Links"]) > 0:
+            for msg in feed_response["Links"]:
+                paylink_feed.paylink(msg)
             response = requests.get(url=url, headers=self.client.headers())
             if "ApiErrorCode" in response.headers:
                 error = response.json()
                 raise Exception("Error feed : %s" % error)
-            feedResponse = response.json()
+            feed_response = response.json()
 
 
 class PaylinkFeed:

twikey/transaction.py

@@ -17,7 +17,7 @@ def create(self, data):
             raise Exception("Error sending transaction : %s" % error)
         return response.json()["Entries"][0]
 
-    def feed(self, transactionFeed):
+    def feed(self, transaction_feed):
         url = self.client.instance_url("/transaction")
 
         self.client.refreshTokenIfRequired()
@@ -26,15 +26,15 @@ def feed(self, transactionFeed):
         if "ApiErrorCode" in response.headers:
             error = response.json()
             raise Exception("Error feed : %s" % error)
-        feedResponse = response.json()
-        while len(feedResponse["Entries"]) > 0:
-            for msg in feedResponse["Entries"]:
-                transactionFeed.transaction(msg)
+        feed_response = response.json()
+        while len(feed_response["Entries"]) > 0:
+            for msg in feed_response["Entries"]:
+                transaction_feed.transaction(msg)
             response = requests.get(url=url, headers=self.client.headers())
             if "ApiErrorCode" in response.headers:
                 error = response.json()
                 raise Exception("Error creating : %s" % error)
-            feedResponse = response.json()
+            feed_response = response.json()
 
 
 class TransactionFeed: