With our API you can fetch payment transactions from bank accounts, credit and debit cards in near-real time. This tutorial will show you how.
Program strategy
In the most common scenario you will have a lot of your consumers enrolled in your Pentadata profile. In our system, each consumer is a “Person”.
Every person can have zero, one or more bank accounts or cards.
With all that taken into account, a simple strategy for updating your own system with the most recent transactions for all your consumer is to have a script that runs periodically and execute the following simple steps:
- Get a list of all Persons in your Pentadata profile.
- For every such Person, get a list of all accounts (both for banks and cards).
- For every such account, get the most recent transactions.
Simple enough, don’t you think?
The remainder of this tutorial shows how to implement such a script in Python. At the end of it you can find a fully working code template.
As a side note, remember that to add accounts in your Pentadata profile the steps are:
- Add a consumer account with the endpoint POST /persons
- Add a bank account for that consumer with the endpoint POST /accounts
- Repeat for all your users!
Preliminaries
To interact with our API the easiest solution is to get the modular code we provide in our GitHub. The Python version is on this page. If you don’t find a module for your programming language, don’t worry: our API is a very standard HTTP API, so you can just send normal requests. The modules we provide are wrappers around it, which handles the JWT authentication.
In 1-liner you can do
wget https://raw.githubusercontent.com/Pentadata/Pentadata-python/main/pentadata_api.py -O pentadata_api.py
Get all Persons (Consumers)
Here’s a simple function that solves this problem. The only argument is an object of the class you should have cloned from GitHub in the Preliminaries step. If you need to call the API directly, just look up the URL that’s in this code.
def get_persons(penta: PentaApi) -> list: """List all consumers (person) in your profile.""" url = 'https://api.pentadatainc.com/persons' response = penta.get(url) if response.status_code != 200: log.warning(response.text) return None data = response.json() persons = data['persons'] log.debug('Found %s persons', len(persons)) return persons
Get all accounts for one Person (Consumer)
Before getting all transactions for all consumers, let’s solve the simpler problem to get all accounts for just one consumer. Here’s a function to do it.
def get_person_accounts(penta: PentaApi, person_id: int) -> list: """List all accounts (banks and cards) of a person.""" url = f'https://api.pentadatainc.com/persons/{person_id}/accounts' response = penta.get(url) if response.status_code != 200: log.warning(response.text) return None data = response.json() accounts = data['accounts'] log.debug('Found %s accounts for person %s', len(accounts), person_id) return accounts
Get all accounts for all Persons (Consumers)
Now we can use the function from the previous step as part of another one that gets all accounts for all consumers. Here it is.
def get_all_persons_accounts(penta: PentaApi, persons: list) -> dict: """Fetch all accounts for all persons. :return: (dict) Keys are person IDs (int), values are accounts' data (dict). """ final = {} for pers in persons: if pers is None: log.warning('Skipping NULL person') continue pid = pers['id'] accounts = get_person_accounts(penta, pid) final[pid] = accounts return final
Get Transactions from one account
We have one endpoint dedicated to getting transactions from a bank account, credit or debit card. The endpoint accepts several parameters, so be sure to look up the docs.
Here’s a function that implements this call and that we can use in the script we’re writing.
def get_account_transactions(penta: PentaApi, account_id: int, **kwargs) -> list: """Fetch all transactions from an account (card or bank). Check the docs for the optional arguments: https://www.pentadatainc.com/docs/introduction/endpoints-reference/get-accounts-id-transactions/ """ url = f'https://api.pentadatainc.com/accounts/{account_id}/transactions' response = penta.get(url, params=kwargs) if response.status_code != 200: log.warning(response.text) return None data = response.json() trns = data['transactions'] total = data['total'] log.debug('Fetched %s transactions (out of %s) for account %s', len(trns), total, account_id) return trns
Get Transactions from all accounts
Like we did before, let’s use the function in the previous step as part of a more general function that fetches transactions from all accounts you have in your Pentadata profile, that is every account for every consumer.
def get_all_accounts_transactions(penta: PentaApi, person_accounts: dict, **kwargs): """Fetch all transactions for all given accounts (bank and/or card) and yield an iterator for each individual account's list of transactions. :param (person_accounts): (dict) Dictionary as returned by `get_all_persons_accounts`. """ for person_id, accounts in person_accounts.items(): for account in accounts: aid = account['account_id'] transactions = get_account_transactions(penta, aid, **kwargs) yield person_id, aid, transactions
Process the transactions
Surely you must have something in mind about what to do with the transaction data. You may want to display them to each user (make sure you don’t show somebody’s transactions to somebody else!), or you may want to store the data for later processing. Here’s a template function, just to make the whole script work; you should implement your custom logic in it.
def process_new_transactions(person_id: int, account_id: int, data: list): """Implement this function with custom processing logic.""" log.debug('New batch of transactions for account %s (person %s) >> %s', account_id, person_id, data)
Connect the dots
Now we can connect all functions written up to this point and write a generic function that fetches all transactions, for all consumers, from all accounts and process each of them.
Re-using the previous parts this is going to be remarkably simple.
def entrypoint(): """Fetch all transactions for all enrolled persons (consumers) for the past 24 hours. **Note** Timestamps are in UTC. """ from_date = datetime.utcnow() - timedelta(hours=24) from_date = from_date.strftime('%Y%m%d%H%M%S') email = 'YOUR-EMAIL' api_key = 'YOUR-API-KEY' penta = PentaApi(email, api_key) persons = get_persons(penta) log.debug('Enrolled persons: %s', persons) accounts = get_all_persons_accounts(penta, persons) log.debug(accounts) transactions = get_all_accounts_transactions(penta, accounts, fromDate=from_date) for pid, aid, new in transactions: process_new_transactions(pid, aid, new)
A few things worth keeping in mind:
- The code above assumes you want to get data from the past 24 hours. You can simply change that argument if your system must act differently. The endpoint also accepts “toDate” parameter.
- Remember that all timestamps in our systems are UTC.
- Add proper, production-ready error monitoring and handling. Errors can be generated especially if you send a wrong person ID or account ID. Save the logs (like shown in the full template below) and get in touch to solve any issue.
Template Code
Here’s a working template (you must plug-in your credentials), and you can also download it from GitHub.
import logging from datetime import datetime, timedelta import requests from pentadata_api import PentaApi logging.basicConfig(format='%(asctime)s %(message)s', level=logging.DEBUG) log = logging.getLogger(__name__) def get_persons(penta: PentaApi) -> list: """List all consumers (person) in your profile.""" url = 'https://api.pentadatainc.com/persons' response = penta.get(url) if response.status_code != 200: log.warning(response.text) return None data = response.json() persons = data['persons'] log.debug('Found %s persons', len(persons)) return persons def get_person_accounts(penta: PentaApi, person_id: int) -> list: """List all accounts (banks and cards) of a person.""" url = f'https://api.pentadatainc.com/persons/{person_id}/accounts' response = penta.get(url) if response.status_code != 200: log.warning(response.text) return None data = response.json() accounts = data['accounts'] log.debug('Found %s accounts for person %s', len(accounts), person_id) return accounts def get_all_persons_accounts(penta: PentaApi, persons: list) -> dict: """Fetch all accounts for all persons. :return: (dict) Keys are person IDs (int), values are accounts' data (dict). """ final = {} for pers in persons: if pers is None: log.warning('Skipping NULL person') continue pid = pers['id'] accounts = get_person_accounts(penta, pid) final[pid] = accounts return final def get_account_transactions(penta: PentaApi, account_id: int, **kwargs) -> list: """Fetch all transactions from an account (card or bank). Check the docs for the optional arguments: https://www.pentadatainc.com/docs/introduction/endpoints-reference/get-accounts-id-transactions/""" url = f'https://api.pentadatainc.com/accounts/{account_id}/transactions' response = penta.get(url, params=kwargs) if response.status_code != 200: log.warning(response.text) return None data = response.json() trns = data['transactions'] total = data['total'] log.debug('Fetched %s transactions (out of %s) for account %s', len(trns), total, account_id) return trns def get_all_accounts_transactions(penta: PentaApi, person_accounts: dict, **kwargs): """Fetch all transactions for all given accounts (bank and/or card) and yield an iterator for each individual account's list of transactions. :param (person_accounts): (dict) Dictionary as returned by `get_all_persons_accounts`. """ for person_id, accounts in person_accounts.items(): for account in accounts: aid = account['account_id'] transactions = get_account_transactions(penta, aid, **kwargs) yield person_id, aid, transactions def process_new_transactions(person_id: int, account_id: int, data: list): """Implement this function with custom processing logic.""" log.debug('New batch of transactions for account %s (person %s) >> %s', account_id, person_id, data) def entrypoint(): """Fetch all transactions for all enrolled persons (consumers) for the past 24 hours. **Note** Timestamps are in UTC. """ from_date = datetime.utcnow() - timedelta(hours=24) from_date = from_date.strftime('%Y%m%d%H%M%S') email = 'YOUR-EMAIL' api_key = 'YOUR-API-KEY' penta = PentaApi(email, api_key) persons = get_persons(penta) log.debug('Enrolled persons: %s', persons) accounts = get_all_persons_accounts(penta, persons) log.debug(accounts) transactions = get_all_accounts_transactions(penta, accounts, fromDate=from_date) for pid, aid, new in transactions: process_new_transactions(pid, aid, new) if __name__ == '__main__': """Run the script entrypoint.""" try: entrypoint() except: log.exception('Error.') # Save the logs and report the error to us else: log.debug('Done.')