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:

1. Get a list of all Persons in your Pentadata profile.
2. For every such Person, get a list of all accounts (both for banks and cards).
3. 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:

1. Add a consumer account with the endpoint POST /persons
2. Add a bank account for that consumer with the endpoint POST /accounts
3. 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.')