Updating python lib to function with Xs2a and adding Ponto Connect

ibanity/Flatten.py

@@ -0,0 +1,36 @@
+def flatten_json(to_flatten): 
+    out = {} 
+
+    def flatten(flat, name =''): 
+        if type(flat) is dict:  
+            for a in flat: 
+                flatten(flat[a], a) 
+        elif type(flat) is list:  
+            i = 0
+            for a in flat:                 
+                flatten(a, name) 
+                i += 1
+        else:
+            if name in out:
+                None
+            else: 
+                out[name] = flat
+    flatten(to_flatten) 
+    return out 
+
+def flatten_json_ponto(to_flatten): 
+    out = {} 
+
+    def flatten_ponto(flat, name =''): 
+        if type(flat) is dict:     
+            for a in flat: 
+                flatten_ponto(flat[a], a) 
+        elif type(flat) is list: 
+            i = 0
+            for a in flat:                 
+                flatten_ponto(a, name) 
+                i += 1
+        else:
+                out[name] = flat 
+    flatten_ponto(to_flatten) 
+    return out 

ibanity/Ibanity.py

@@ -4,6 +4,7 @@
 from json import loads, dumps
 
 import ibanity.Error
+import ibanity.Flatten
 
 
 class Ibanity:
@@ -16,6 +17,7 @@ def __init__(self, certificate_path, key_path, key_passphrase, api_host, scheme=
         self.api_host = api_host
         self.base_uri = self.scheme + "://" + self.api_host
         self.schema = None
+        self.ponto_schema = None
         self.__class__.client = self
 
     @property
@@ -24,6 +26,12 @@ def api_schema(self):
             self.schema = self.get(self.base_uri, {}, None)["links"]
         return self.schema
 
+    @property
+    def api_schema_ponto(self):
+        if self.ponto_schema is None:
+            self.ponto_schema = self.get(self.base_uri + "/ponto-connect", {}, None)["links"]
+        return self.ponto_schema
+
     def get(self, uri, params, customer_access_token):
         return self.execute("GET", uri, {}, params=params, headers=self.__build_headers(customer_access_token))
 
@@ -54,12 +62,14 @@ def execute(self, method, uri, body, params, headers):
 
     @staticmethod
     def __build_headers(customer_access_token):
-        authorization = ""
-        if customer_access_token:
-            authorization = "Bearer " + str(customer_access_token)
-
+        if (customer_access_token == None):
+            return {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "Authorization": ""
+            }    
         return {
             "Content-Type": "application/json",
             "Accept": "application/json",
-            "Authorization": authorization
+            "Authorization": customer_access_token
         }

ibanity/__init__.py

@@ -1,12 +1,24 @@
 from ibanity.Ibanity import Ibanity
 from ibanity.Error import Error
-import ibanity.api.FinancialInstitution
-import ibanity.api.CustomerAccessToken
-import ibanity.api.AccountInformationRequest
-import ibanity.api.Account
-import ibanity.api.Transaction
-import ibanity.api.PaymentInitiationRequest
-import ibanity.api.SandboxAccount
-import ibanity.api.SandboxUser
+import ibanity.api.Xs2a.FinancialInstitution
+import ibanity.api.Xs2a.Customer
+import ibanity.api.Xs2a.CustomerAccessToken
+import ibanity.api.Xs2a.Synchronization
+import ibanity.api.Xs2a.AccountInformationAccessRequest
+import ibanity.api.Xs2a.AccountInformationAccessRequestAuthorization
+import ibanity.api.Xs2a.Account
+import ibanity.api.Xs2a.Transaction
+import ibanity.api.Xs2a.Holding
+import ibanity.api.Xs2a.PaymentInitiationRequest
+import ibanity.api.Xs2a.PaymentInitiationRequestAuthorization
+import ibanity.api.Sandbox.FinancialInstitutionHolding
+import ibanity.api.Sandbox.FinancialInstitutionAccount
+import ibanity.api.Sandbox.FinancialInstitutionUser
 import ibanity.api.SandboxFinancialInstitution
-import ibanity.api.SandboxTransaction
+import ibanity.api.Sandbox.FinancialInstitutionTransaction
+import ibanity.api.PontoConnect.Account
+import ibanity.api.PontoConnect.FinancialInstitution
+import ibanity.api.PontoConnect.Payment
+import ibanity.api.PontoConnect.Synchronization
+import ibanity.api.PontoConnect.Token
+import ibanity.api.PontoConnect.Transaction

ibanity/api/PontoConnect/Account.py

@@ -0,0 +1,23 @@
+from collections import namedtuple
+from ibanity import Ibanity
+from ibanity.Flatten import flatten_json_ponto as flatten_json
+
+
+def get_list(access_token):
+    uri = Ibanity.client.api_schema_ponto["accounts"] \
+        .replace("{accountId}", "")
+    response = Ibanity.client.get(uri, {}, "Bearer " + str(access_token))
+    return list(
+        map(
+            lambda account:
+            flatten_json(account), response["data"]
+        )
+    )
+
+
+def find(id, access_token):
+    uri = Ibanity.client.api_schema_ponto["accounts"] \
+        .replace("{accountId}", id)
+    response = Ibanity.client.get(uri, {}, "Bearer " + str(access_token))
+    return flatten_json(response["data"])
+

ibanity/api/PontoConnect/FinancialInstitution.py

@@ -0,0 +1,26 @@
+from collections import namedtuple
+from ibanity import Ibanity
+from ibanity.Flatten import flatten_json
+
+
+def get_list(access_token = None):
+    uri = Ibanity.client.api_schema_ponto["financialInstitutions"] \
+    .replace("{financialInstitutionId}", "")
+    if (access_token == None):
+        response = Ibanity.client.get(uri, {}, None)
+    else:
+        response = Ibanity.client.get(uri, {}, "Bearer " + str(access_token))
+    return list(
+        map(
+            lambda financial_institution:
+            flatten_json(financial_institution), response["data"]
+        )
+    )
+
+
+def find(id):
+    uri = Ibanity.client.api_schema_ponto["financialInstitutions"] \
+        .replace("{financialInstitutionId}", id)
+    response = Ibanity.client.get(uri, {}, None)
+    return flatten_json(response["data"])
+

ibanity/api/PontoConnect/Payment.py

@@ -0,0 +1,34 @@
+from collections import namedtuple
+from ibanity import Ibanity
+from ibanity.Flatten import flatten_json
+
+
+def create(account_id, attributes, access_token):
+    uri = Ibanity.client.api_schema_ponto["account"]["payments"] \
+        .replace("{accountId}", account_id) \
+	    .replace("{paymentId}", "")
+    body = {
+        "data": {
+            "type": "payment",
+            "attributes": attributes
+            }
+        }
+    response = Ibanity.client.post(uri, body, {}, "Bearer " + str(access_token))
+    return flatten_json(response["data"])
+
+
+def find(account_id, id, access_token):
+    uri = Ibanity.client.api_schema_ponto["account"]["payments"] \
+        .replace("{accountId}", account_id) \
+        .replace("{paymentId}", id)
+    response = Ibanity.client.get(uri, {}, "Bearer " + str(access_token))
+    return flatten_json(response["data"])
+
+
+def delete(account_id, id, access_token):
+    uri = Ibanity.client.api_schema_ponto["account"]["payments"] \
+        .replace("{accountId}", account_id) \
+        .replace("{paymentId}", id)
+    response = Ibanity.client.delete(uri, {}, "Bearer " + str(access_token))
+    return flatten_json(response["data"])
+

ibanity/api/PontoConnect/Synchronization.py

@@ -0,0 +1,24 @@
+from collections import namedtuple
+from ibanity import Ibanity
+from ibanity.Flatten import flatten_json
+
+
+def create(attributes, access_token):
+    uri = Ibanity.client.api_schema_ponto["synchronizations"] \
+	    .replace("{synchronizationId}", "")
+    body = {
+        "data": {
+            "type": "synchronization",
+            "attributes": attributes
+            }
+        }
+    response = Ibanity.client.post(uri, body, {}, "Bearer " + str(access_token))
+    return flatten_json(response["data"])
+
+
+def find(id, access_token):
+    uri = Ibanity.client.api_schema_ponto["synchronizations"] \
+        .replace("{synchronizationId}", id)
+    response = Ibanity.client.get(uri, {}, "Bearer " + str(access_token))
+    return flatten_json(response["data"])
+

ibanity/api/PontoConnect/Token.py

@@ -0,0 +1,28 @@
+from collections import namedtuple
+from ibanity import Ibanity
+import requests
+from ibanity.Flatten import flatten_json
+
+
+def create(authorization_code, code_verifier, redirect_uri, client_id, authorization):
+    uri = Ibanity.client.api_schema_ponto["oauth2"]["token"]
+    body = {
+        "grant_type": "authorization_code",
+        "code": authorization_code,
+        "client_id": client_id,
+        "redirect_uri": redirect_uri,
+        "code_verifier": code_verifier
+        }
+    response = Ibanity.client.post(uri, body, {}, "Basic " + str(authorization))
+    return flatten_json(response)
+
+
+def create_from_refresh_token(refresh_token, client_id, authorization):
+    uri = Ibanity.client.api_schema_ponto["oauth2"]["token"]
+    body = {
+        "grant_type": "refresh_token",
+        "refresh_token": refresh_token,
+        "client_id": client_id,
+        }
+    response = Ibanity.client.post(uri, body, {}, "Basic " + str(authorization))
+    return flatten_json(response)

ibanity/api/PontoConnect/Transaction.py

@@ -0,0 +1,25 @@
+from collections import namedtuple
+from ibanity import Ibanity
+from ibanity.Flatten import flatten_json_ponto as flatten_json
+
+
+def get_list(account_id, access_token):
+    uri = Ibanity.client.api_schema_ponto["account"]["transactions"] \
+	    .replace("{accountId}", account_id) \
+		.replace("{transactionId}", "")
+    response = Ibanity.client.get(uri, {}, "Bearer " + str(access_token))
+    return list(
+        map(
+            lambda transaction:
+            flatten_json(transaction), response["data"]
+        )
+    )
+
+
+def find(account_id, id, access_token):
+    uri = Ibanity.client.api_schema_ponto["account"]["transactions"] \
+        .replace("{accountId}", account_id) \
+		.replace("{transactionId}", id)
+    response = Ibanity.client.get(uri, {}, "Bearer " + str(access_token))
+    return flatten_json(response["data"])
+

ibanity/api/Sandbox/FinancialInstitutionAccount.py

@@ -0,0 +1,52 @@
+from collections import namedtuple
+from ibanity import Ibanity
+from ibanity.Flatten import flatten_json
+
+
+def get_list_for_financial_institution(financial_institution_id, financial_institution_user_id, params={}):
+    uri = Ibanity.client.api_schema["sandbox"]["financialInstitution"]["financialInstitutionAccounts"] \
+        .replace("{financialInstitutionId}", financial_institution_id) \
+        .replace("{financialInstitutionUserId}", financial_institution_user_id) \
+        .replace("{financialInstitutionAccountId}", "")
+    response = Ibanity.client.get(uri, params, None)
+    print(response)
+    return list(
+        map(
+            lambda account:
+            flatten_json(account), response["data"]
+        )
+    )
+
+
+def create(financial_institution_id, financial_institution_user_id, attributes):
+    uri = Ibanity.client.api_schema["sandbox"]["financialInstitution"]["financialInstitutionAccounts"] \
+        .replace("{financialInstitutionId}", financial_institution_id) \
+        .replace("{financialInstitutionUserId}", financial_institution_user_id) \
+        .replace("{financialInstitutionAccountId}", "")
+    body = {
+        "data": {
+            "type": "financialInstitutionAccount",
+            "attributes": attributes
+        }
+    }
+    response = Ibanity.client.post(uri, body, {}, None)
+    return flatten_json(response["data"])
+
+
+def delete(financial_institution_id, financial_institution_user_id, id):
+    uri = Ibanity.client.api_schema["sandbox"]["financialInstitution"]["financialInstitutionAccounts"] \
+        .replace("{financialInstitutionId}", financial_institution_id) \
+        .replace("{financialInstitutionUserId}", financial_institution_user_id) \
+        .replace("{financialInstitutionAccountId}", id)
+    response = Ibanity.client.delete(uri, {}, None)
+    return flatten_json(response["data"])
+
+
+def find(financial_institution_id, financial_institution_user_id, id):
+    uri = Ibanity.client.api_schema["sandbox"]["financialInstitution"]["financialInstitutionAccounts"] \
+        .replace("{financialInstitutionId}", financial_institution_id) \
+        .replace("{financialInstitutionUserId}", financial_institution_user_id) \
+        .replace("{financialInstitutionAccountId}", id)
+    response = Ibanity.client.get(uri, {}, None)
+    return flatten_json(response["data"])
+