Automate Customer Rewards using Python, Postgresql and Africas Talking

A while back me, and a friend (shout out to Nyambura for this awesome idea) were talking about customer reward programs and promotions. The discussion was mostly about how currently organizations tend to do things manually, yet most already tend to have some sort of digital system in place with a database backend storing all their records.

In this article we will be part of the marketing team of our fictional chain of ZOO supermarkets. We are tasked with running a customer rewards program to celebrate our 1st year anniversary. Specifically we take on the challenge of awarding airtime to our customers who will shop above a certain amount during the anniversary period.

There are a few proprietary options including outsourcing this task to a 3rd-Party provider. However, at ZOO we consider ourselves a 'digital-first' franchise thus we decide to develop an in-house solution. The assumption is we have a POS system with a Postgresql database backend. We decide to use Python as our language of choice and the awesome AfricasTalking airtime api

Getting Prepared

To follow along this post and code the same features. You're going to need a few things:

• Python and pip (I am currently using 3.9.2) Any version above 3.5 should work.

• An Africas Talking account.

  • Api Key and username from your account. Create an app and take note of the api key.
    • Additionally, you will need to request them to enable airtime access for your account. Email their airtime team for further clarification

  Once you've got the above sorted :

  • Create a new directory and change into it.
    

    
               mkdir customer-rewards 
               cd customer-rewards
    
    

    • Create a new virtual environment for the project or activate the previous one.
    • Using python package manager(pip), install: africastalking python sdk, python-dotenv library, sqlachemy, psycopg2 and sqlachemy-utils libraries.
    • Save the installed libraries in a requirements.txt file

  
         python -m venv .
         source bin/activate
         pip install africastalking python-dotenv sqlachemy sqlachemy-utils psycopg2  
         pip freeze > requirements.txt
  
  

As mentioned above we are using Postgresql as our database of choice hence we need a library to interface with the database, psycopg2 is a good option although there are others. Although not necessary we'll be making use of SqlAlchemy as our Object Relation Mapper(ORM). This allows us to use python objects (classes, functions) to make transactions instead of raw SQL.

Install Postgresql database to simulate a backend with customer details. Depending on which platform you code on, you could do it natively on your system. Personally I am using docker as it is easy to manage containers and prevents my system from being cluttered. This article is an awesome resource on how to get Postgresql and pgadmin4 installed as containers. Also you could use a database administration tool like pgadmin4 or DBeaver for a UI to interface with Postgres server.

Alternatively, check the finished code on Github

Config For Days

As per our current assumption we have a database of customer data and in production that should definitely be the case with backups in place. However, for this showcase we'll create a sample database and tables. This will make it easier to explain the flow of the program.

This step is entirely optional, You could just as easily create a db manually using a database client like dbeaver. However, using code is the optimal route. Let's create a .env file to hold all of our environment variables that we dont want to expose to everyone. Enter the following changing the placeholders with the proper credentials.

    touch .env

# .env 
postgres_db=enter your database here
postgres_host=localhost
postgres_port=5432
postgres_user=enter postgres_user
postgres_password=enter postgres_user password
at_username=enter your at_username
at_api_key=enter your api_key
phone_number=07xxxxxxxx

We'll create a config.py file to hold all of our configurations. Inside our config file, lets go ahead and import all of our required libraries.

    #config.py
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv

The block of code above:

1. imports the create_engine function to connect with the database as well as Metadata object.
2. We then import the declarative_base factory function that constructs a base class for declarative class definitions.
3. We proceed to import the sessionmaker class that is normally used to create a top level Session configuration which can then be used throughout an application without the need to repeat the configurational arguments.
4. We import the os lib to enable access to environment variables.
5. We also import the load_dotenv function to load values from our .env file.

    # config.py 
load_dotenv()
db = os.getenv('postgres_db')
db_host = os.getenv('postgres_host')
db_port = os.getenv('postgres_port')
db_user = os.getenv('postgres_user')
db_password = os.getenv('postgres_password')

The code block above utilizes the load_dotenv function to get values from our .env file. We proceed to create variables to hold our specific environment values.

We now go into Database specific configuration. The code below:

• Creates an instance of the create_engine function passing required parameters in relation to the db.
• We also create instances of MetaData, declarative_base and Session functions for use across our program.

# config.py
engine = create_engine(f"postgresql+psycopg2://{db_user}:{db_password}@{db_host}:{db_port}/{db}", echo=True)
meta = MetaData()
Base = declarative_base()
Session = sessionmaker(bind=engine, autoflush=True)
session = Session()

More Customers More Data

Create a new python script named database_insert.py this will hold all the code for inserting data for our customers and sales transactions.
Here we will define how our data will be stored inside our database, this will also serve a good primer on using sqlachemy for CRUD (Create Read Update Delete) operations.

Inside database_insert.py add the following code:

# database_insert.py
from sqlalchemy import Column, Integer, String, Unicode, ForeignKey, DateTime
import sqlalchemy as sa
from sqlalchemy_utils import PhoneNumber
import datetime as dt
import os
from dotenv import load_dotenv
from config import engine, Base, session

The above code imports all of our required libraries and the pre-defined configuration
from config.py for use when accessing our database.

We'll now proceed to define our classes for how data will be stored.

# database_insert.py

phone_number = '07XXXXXXXX'


class Customers(Base):
    __tablename__ = 'customers'
    customer_id = Column(Integer, primary_key=True)
    customer_first_name = Column(String)
    customer_last_name = Column(String)
    _phonenumber = Column(Unicode(20))
    country_code = Column(Unicode(8))
    phone_number = sa.orm.composite(
        PhoneNumber, _phonenumber, country_code
    )


class Sales_Transaction(Base):
    __tablename__ = 'sales_transaction'
    transaction_id = Column(Integer, autoincrement=True, primary_key=True)
    customer_id = Column(Integer, ForeignKey("customers.customer_id"))
    transaction_price = Column(Integer)
    transaction_date = Column(DateTime(timezone='EAT'))

In the above snippet we create two classes: Customers and Sales_Transaction both of which inherit from the Base class from sqlachemy. This enables us easily map our models to our tables. In case you want more information, the documentation
provides great deep dive.

# database_insert.py 
Base.metadata.create_all(engine)

temp = Customers(phone_number=PhoneNumber(f'{phone_number}', 'KE'))
c1 = Customers(customer_id=2, customer_first_name='babygirl',
               customer_last_name='nyambura', _phonenumber=temp.phone_number.e164)
c2 = Customers(customer_id=3, customer_first_name='zoo',
               customer_last_name='mwaura',
               _phonenumber=temp.phone_number.e164)
s2 = Sales_Transaction(customer_id=c1.customer_id,
                       transaction_price=5000, transaction_date=dt.date(2021, 4, 12))
s3 = Sales_Transaction(customer_id=c2.customer_id, transaction_price=2700,
                       transaction_date=dt.date(2021, 4, 15))

try:
    session.add_all([c1, c2])
    session.add_all([s2, s3])
    session.commit()
except Exception as e:
    print(f"We have a problem Houston: {e}")

session.close()

The above snippet makes use of the Base class to create all of our tables as specified in our classes. We proceed to create a temp variable to hold our unformatted phone number. We make use of the PhoneNumber function from the sqlachemy_utils library to set the correct type of our phone_number. We then add variables to insert test data into the database. The data provided for test purposes, use your own as required. We created a Foreign Key field in our Sales_Transaction table inorder to have a relationship between our Customers table and Sales table. Each transaction should have a customer id to easily identify which customer made which transaction.

The Database should resemble the diagram below:

The Search Begins!

In this section we will write queries against our database for customers who we will
subsequently reward. Most of the time you would have to do this in native SQL code,
however, we have an ORM in place thus we will write python code inorder to query our
database.

Inside our working folder, create a python file named customer_search.py. This file
just as the name implies will be used to query our customer data. Our original task was to reward customers during our anniversary period who have shopped above a certain amount. We'll set our shopping period to 12th March 2021 to 19th March 2021 and shopping amount to equal to and above kes 2500.

# customer_search.py

from datetime import date
from config import session
from database_insert import Customers, Sales_Transaction

class CustomerQuery:
    def customer_query():
        start_dt = date(2021, 4, 12)
        end_dt = date(2021, 4, 19)
        print(start_dt)
        sq = session.query(
            Sales_Transaction.transaction_price, Customers.customer_first_name,
            Customers.customer_last_name, Customers.phone_number).
        join(Customers)
    customer_data = []
    for a, b, c, d in sq.filter(Sales_Transaction.transaction_price > 2500)
        .filter(Sales_Transaction.transaction_date >= start_dt)
        .filter(Sales_Transaction.transaction_date <= end_dt):
    cl = [a, b, c, d.e164]

    customer_data.extend([cl])

return customer_data


print(CustomerQuery.customer_query())

In our code block above we import date from datetime library, We also get the session
fom our config file. We also get the Customers and Sales_Transaction classes from the
database_insert.py. We create a CustomerQuery class, define a customer_query function to hold all of our queries. We define a start (start_dt) and end(end_dt) date variables inorder to specify the period in between(the anniversary period).

We then create a query(sq) on our session. We query the Sales_Transaction class and Customers class, we then create an empty customer_data list that we append our query results. Since we converted our phone_number to PhoneNumber class, we can easily get in e.164 format as required by africastalking by calling the .e164 method. We then return our customer_data list containing all of our required data.

Everyone gets a Reward!

Finally we get to rewarding our esteemed shoppers! Create a file customer_rewards.py.

Inside the file add the following code:

# customer_rewards.py 

import africastalking as at
from dotenv import load_dotenv
import os

from customer_search import CustomerQuery

load_dotenv()
# get the environment values from the .env file
at_username = os.getenv('at_username')
at_api_key = os.getenv('at_api_key')

at.initialize(at_username, at_api_key)
airtime = at.Airtime
account = at.Application

print(account.fetch_application_data())

def customer_rewards():
    # Set The 3-Letter ISO currency code and the amount
    amount = "250"
    currency_code = "KES"
    for n in CustomerQuery.customer_query():
        print(n[1], n[2], n[-1])
        numbers.append(n[-1])
    print(numbers)
    for number in numbers:
        try:
            response = airtime.send(phone_number=number, amount=amount, currency_code=currency_code)
            print(response)
        except Exception as e:
            print(f"Encountered an error while sending airtime. More error details below\n {e}")

customer_rewards()

In the block above, we import the africas talking python library as well load_dotenv function to get our africas talking credentials. We proceed to assign variables to our api key and username, we then initialize the africas talking client by passing our values as arguments. We then assign the airtime class to a variable, and the account as well. We fetch our current account balance using account.fetch_application_data() function.

We define a customer_rewards function, inside the function we set an amount, currency code variables. These will hold our airtime amount and three-digit ISO code required by africas talking. Read further in the documentation. We loop through the customer data and get each number to be rewarded. We then insert a try-catch block we then attempt to send airtime to each customer using the airtime.send function. We then call our customer_rewards() function. If all went well, your customers should receive their airtime. Voila! We have successfully developed an automated method to effeciently and quickly reward our loyal customers. There is of course room for improvement:

1. Adding logging to keep track of customers awarded.
2. Connecting to a frontend application to visualize the total amount of airtime sent. Alternatively crafting an API around the rewards program.

But this was more of a demo project to illustrate the viability of the idea.
One feature I found to be amazing was, the Africas Talking airtime api would return an error whenever I sent airtime to the same number more than once in a span of 5 minutes.
This would be life-saver when you have numerous customers to reward, it would prevent double or triple rewards without any additional code on our end.

If you have any question or comments. Let me know in the comments, or on Twitter.
Happy Coding.

Ps: Massive shoutout to Anthony Limo, Anne from Africas talking airtime and by extension the airtime team. The onboarding and KYC process was quick and painless.

Buy me a coffee