Introduction
This is a practice mini project for me to get hold of API, sqlite3 and regex. The objective is to capture contact information and store into a sqlite3 database file – contacts.db. This is not a full fledged web application for storing contacts however can be used as a based for additional features.
Information that needs to be captured after contact registration:
- last_name, this is mandatory
- first_name, this is mandatory
- email_address, this is optional
- location_address, this is optional
- contact_number, this is optional
Design API endpoint
Use Postman to design the url for GET, DELETE and POST.
- Get contact list, this gets all contacts in the database
- Get single contact by contact id
- Post contact information
- Delete contact by contact_id
These are the endpoints which I created from Postman.
This is the POST method of endpoint
/register
This is the GET method to get all contacts.
This is the GET method to get a single contact based on contact_id.
This is the DELETE method to delete a single contact based on contact_id.
Contacts.py description
The endpoints use three resources, and each resources is a class. To use the API resources flask_RESTful.Resource is imported. Another import is a flask_RESTful.reqparse(), the reqparse is a request parser that checks the json fields for required data type or if the field is required.
I have also imported sqlite3 as it is required to store the data from post request, and also to retrieve data from get request.
RE is imported to use to match email regex pattern.
A moment to explain the regular expression pattern before showing the entire contacts.py code.
"\S+@[a-zA-Z0-9]+\.[a-zA-Z]{2,5}"
This regex can be tested here.
\S+, matches non-whitespace characters one or more times, non-whitespace can be any characters including special characters.@[a-zA-Z0-9]+\.,@appears once, the domain name can be any valid alphanumeric, the dot sign is wildcard in regex hence has to be escaped by backward slash.[a-zA-Z]{2,5}, this matches valid root for the domain, minimum must be two characters to match root such ascn,sg,vn,ca,usand at most is 5 characters this is to match the root such aslocal
Now the below is the entire contacts.py with comments within the python code.
import sqlite3
from flask_restful import reqparse, Resource
import re
# Register contact resource, this resource api endpoint is /register uri
class Register(Resource):
# Create RequestParser object
parser = reqparse.RequestParser()
# last_name field is mandatory, and is string.
parser.add_argument('last_name',
type=str,
required=True,
help='This field cannot be left blank.'
)
# first_name field is mandatory, and is string.
parser.add_argument('first_name',
type=str,
required=True,
help='This field cannot be left blank.')
# email_address field is a string
parser.add_argument('email_address',
type=str,
help="This field is a string.")
# location_address field is a string
parser.add_argument('location_address',
type=str,
help='This field is a string')
# contact_number field is a string
parser.add_argument('contact_number',
type=str,
help='This field is a string of numbers.')
# database table name
TABLE_NAME = 'contacts'
# POST method
def post(self):
# match email pattern
email_pattern = "\S+@[a-zA-Z0-9]+\.[a-zA-Z]{2,5}"
regex = re.compile(email_pattern)
query = "INSERT INTO {} VALUES (NULL, ?, ?, ?, ?, ?, ?)".format(Register.TABLE_NAME)
# JSON data is parsed and store to data.
data = Register.parser.parse_args()
# combine first and last name and store in fullname column in contacts table.
fullname = data['first_name'] + " " + data['last_name']
# check valid email address format, if valid then keep the data else make it to null.
data['email_address'] = data['email_address'] if regex.match(data['email_address']) else None
# prepare the list to be passed into the query.
# the use of get method can pass None if the data is not in the json field.
# get method is for dictionary only, since json is a dictionary as well.
data_list = [
data.get('last_name', None),
data.get('first_name', None),
data.get('email_address', None),
data.get('location_address', None),
data.get('contact_number', None)
]
'''
All sqlite3 db codes use the below 5 steps
'''
# Step 1: Connect db file.
db = sqlite3.connect('contacts.db')
# Step 2: create cursor object
cursor = db.cursor()
# Step 3: execute query
cursor.execute(query, (fullname, *data_list))
# Step 4: commit the changes.
db.commit()
# Step 5: close the db.
db.close()
# Message back to requester, 201 is accepted.
return {"message": "data registered."}, 201
# Get all contacts in a list
class ContactList(Resource):
# GET method
def get(self):
# initialize the contact list.
contact_list = []
# Get all data from contacts table.
query = "SELECT * FROM {}".format(Register.TABLE_NAME)
'''
No changes to database, hence no commit is needed.
The database code only require 4 steps.
Step 1: Connect to database.
Step 2: Create cursor object.
Step 3: Execute the SELECT query
Step 4: Close the database.
'''
db = sqlite3.connect('contacts.db')
cursor = db.cursor()
# The returned result is a list of rows.
result = cursor.execute(query)
# Iterate through the list, for each row.
for row in result:
# Convert the row into dictionary
# contact_list collects dictionary, each dictionary is a contact's information.
contact_list.append({"id": row[0],
"full_name": row[1],
"last_name": row[2],
"first_name": row[3],
"email_address": row[4],
"location_address": row[5],
"contact_number": row[6]})
db.close()
return {"contacts": contact_list}, 200
# Search single contact by id.
class Contact(Resource):
@classmethod
def find_by_id(cls, contact_id):
query = "SELECT * FROM {} WHERE contact_id=?".format(Register.TABLE_NAME)
db = sqlite3.connect('contacts.db')
cursor = db.cursor()
# query variable is in tuple, (contact_id,) is to tell python this is a tuple.
cursor.execute(query, (contact_id,))
# contact_id has a PRIMARY KEY constraint in the database,
# PRIMARY KEY must be unique hence use fetchone()
result = cursor.fetchone()
# If result is not None i.e. if result exists, else this method returns "None"
# The result omits contact_id, since it is already searched by contact_id.
if result:
return {"fullname": result[1],
"last_name": result[2],
"first_name": result[3],
"email_address": result[4],
"location_address": result[5],
"contact_number": result[6]}
# GET method
def get(self, contact_id):
# if contact_id is found in contacts return the result with 200 ok.
found = self.find_by_id(contact_id)
if found:
return found, 200
# if contact_id is Nonetype, returns not found 404, if this is not treated 500 internal server error is raised.
else:
return {"message": "contact information not found"}, 404
# delete a contact base on contact_id.
def delete(self, contact_id):
query = "DELETE FROM {} WHERE contact_id=?".format(Register.TABLE_NAME)
db = sqlite3.connect('contacts.db')
cursor = db.cursor()
# if the contact_id is found in database delete the contact.
if self.find_by_id(contact_id):
cursor.execute(query, (contact_id,))
db.commit()
db.close()
return {"message": "Contact deleted."}, 201
# if contact_id does not exist in database, raise a bad request 400.
else:
return {"message": "contact id not found"}, 400
The following is the code to test the contacts.py.
import sqlite3
from flask import Flask
from flask_restful import Api
from contacts import Register, ContactList, Contact
'''
Create contacts table for contacts database (contacts.db).
If the file contacts.db does not exist create one.
There are 7 columns:
1. contact_id, this is primary key hence is unique, will be incremented automatically.
2. fullname, this is the concatenation of last and first name.
3. last_name, this column stores the last name.
4. first_name, this column stores the first name.
5. email_address, this column stores the email address.
6. location_address, this column stores the physical address.
7. contact_number, this column stores the contact number.
'''
def create_table():
query = "CREATE TABLE IF NOT EXISTS contacts (contact_id INTEGER PRIMARY KEY," \
"fullname TEXT," \
"last_name TEXT," \
"first_name TEXT," \
"email_address TEXT," \
"location_address TEXT," \
"contact_number)"
db = sqlite3.connect('contacts.db')
cursor = db.cursor()
cursor.execute(query)
db.commit()
db.close()
# create the web app.
app = Flask(__name__)
# create the api object.
api = Api(app)
# testing starts here.
if __name__ == '__main__':
create_table()
'''
There are three endpoints based on Postman design.
{{url}}:5000/register - Post new contact
{{url}}:5000/contacts - Get all available contacts
{{url}}:5000/contact/ - Get specific single contact based on contact_id.
corresponds with the contact_id of the get method of class Contact.
'''
api.add_resource(Register, '/register')
api.add_resource(ContactList, '/contacts')
api.add_resource(Contact, '/contact/')
app.run(debug=True)
The third api.add_resource(Contact, '/contact/') was truncated by wordpress as wordpress thought I was putting in html tags.
This is the actual main:
Demonstration
Run the app, in the production webserver uWSGi will be used to run this app on user behalf.
Add the first entry.
Second entry added.
Get all contacts.
Get a single contact by id.
Get a non-existent contact.
Delete a contact that does not exist.
Delete a valid contact.
cyruslab 