Python is one of the most popular programming language used widely in data science, machine learning and data integration scenarios.
Flask is a micro web framework for Python. It is lightweight and easy to use, making it a great choice for building small, simple web applications and APIs.
Oracle Database is the most popular relational, multi-model database in the DB-engine ranking for years now. Oracle is used by some of the largest companies in the world who need a robust, performant, highly secure and scalable database. It is widely used for large-scale enterprise applications, and it supports a wide range of data types and programming languages.
In this article we’ll be combining the power of Python3 and Oracle Database using Flask web framework to create robust RESTful API’s.
The demo in this article was deployed with Python 3.6.5 on Oracle Linux 7 running on Oracle Cloud Infrastructure (OCI) VM
Let’s get started
1. Install Python 3 and the flask , cx_Oracle, Jinga2 & six packages on Oracle Linux 7
First pre-requisite is to ensure your instance has Python3 installed along with the Python packages. We will also install the command-line browser links to test the API using a html form.
sudo yum install python36
sudo yum install links
sudo pip3 install flask
sudo pip3 install six
sudo pip3 install Jinja2
sudo pip3 install cx_Oracle
python3 --version2. Generate Self-signed certificates and firewall rules
As we are creating a secure web server ensure you need SSL certificates. In this example for demo purposes we are creating self-signed certificates but in a production scenario you should have SSL certificates issued from a third party authority.
sudo yum install openssl
mkdir ~/ssl-certs
openssl genpkey -algorithm RSA -out ~/ssl-certs/key.pem
openssl req -new -x509 -key ~/ssl-certs/key.pem -out ~/ssl-certs/cert.pem
chmod +r cert.pem key.pem
sudo firewall-cmd --permanent --add-port=4443/tcp3. Deploy the Oracle Table, Sequence and Trigger
The example uses a simple table called employees in the hr schema. A sequence is used to auto-increment the id of the employee with a BEFORE INSERT trigger. You can also use an identity column for performing the auto-increment on id column.
CREATE SEQUENCE employees_seq;
CREATE TABLE employees (
id NUMBER PRIMARY KEY,
name VARCHAR2(255) NOT NULL,
email VARCHAR2(255) NOT NULL,
department VARCHAR2(255) NOT NULL
);
CREATE OR REPLACE TRIGGER employees_bir
BEFORE INSERT ON employees
FOR EACH ROW
BEGIN
SELECT employees_seq.NEXTVAL
INTO :NEW.id
FROM DUAL;
/
SELECT * FROM user_errors;4. Main Python code
This is the core python3 code to create a web server with Flask and add the routes for the API’s on the employees table in Oracle.
There are 6 functions in the script, each performing a REST call for GET, PUT, POST and DELETE operations. I’ve included an additional function to add an employee using html form via render_template
GET ALL - def get_employees():
/* Fetches all the employees from Oracle table eg : curl https://10.180.1.21:4443/api/employees -k */
GET EMP - def get_employee(id):
/* Fetches a single employee using the id and throws error 404 if employee not found eg : curl https://10.180.1.21:4443/api/employees/1 -k */
ADD EMP - def add_employee():
/* Adds an employee to the Oracle table and throws error if the same email already exists eg : curl -X POST https://10.180.1.21:4443/api/employees -d '{"name":"Shadab M","email":"shadabm@example.com","department":"IT"}' -H "Content-Type: application/json" -k */
UPDATE EMP - def update_employee(id):
/* Updates an employee details but throws 404 error if employee id does not exist eg : curl -X PUT https://10.180.1.21:4443/api/employees/1 -d '{"name":"Michael","email":"michaelm@example.com","department":"IT"}' -H "Content-Type: application/json" -k */
DELETE EMP - def delete_employee(id):
/* Deletes an employee but throws 404 error if employee not found */
GET EMP BY EMAIL - def search_employee(email):
/* Search an employee by their email address, if employee email is not found it returns 404 error eg :curl https://10.180.1.21:4443/api/employees/search/shadabm@example.com -k */
ADD EMP USING HTML FORM - def add_employee_form():
/* Add an employee using an html form with the Jinga2 flask render_template. The render_template function is used to render an HTML template called "add_employee.html" that contains the form. */
import os
import flask
from flask import Flask, jsonify, request
from flask import render_template
from flask_cors import CORS
import cx_Oracle
app = Flask(__name__, template_folder='/home/opc')
CORS(app)
# Connect to the Oracle database
con = cx_Oracle.connect('hr', '**********', 'host:port/servicename')
@app.route('/api/employees', methods=['GET'])
def get_employees():
# Fetch all employees from the database
cur = con.cursor()
cur.execute('SELECT * FROM employees')
employees = cur.fetchall()
# Convert the results to a list of dictionaries
employees_list = []
for employee in employees:
employees_list.append({
'id': employee[0],
'name': employee[1],
'email': employee[2],
'department': employee[3]
})
# Return the list of employees in JSON format
return jsonify(employees_list)
@app.route('/api/employees/<int:id>', methods=['GET'])
def get_employee(id):
# Check if the employee with the specified ID exists
cur = con.cursor()
cur.execute("SELECT * FROM employees WHERE id = :id", {'id': id})
employee = cur.fetchone()
if employee is None:
return jsonify({'message': 'Employee not found'}), 404
# Extract the employee data from the database
name = employee[1]
email = employee[2]
department = employee[3]
# Return the employee data
return jsonify({'id': id, 'name': name, 'email': email, 'department': department}), 200
@app.route('/api/employees/search/<string:email>', methods=['GET'])
def search_employee(email):
# Search the employee by email in the database
cur = con.cursor()
cur.execute("SELECT * FROM employees WHERE email = :email", {'email': email})
employee = cur.fetchone()
if employee is None:
return jsonify({'message': 'Employee not found'}), 404
# Extract the employee data from the database
id = employee[0]
name = employee[1]
department = employee[3]
# Return the employee data
return jsonify({'id': id, 'name': name, 'email': email, 'department': department}), 200
@app.route('/api/add_employee', methods=['GET', 'POST'])
def add_employee_form():
if request.method == 'POST':
# Extract the employee data from the form
name = request.form['name']
email = request.form['email']
department = request.form['department']
# Insert the employee into the database
cur = con.cursor()
cur.execute("INSERT INTO employees (name, email, department) VALUES (:name, :email, :department)",
{'name': name, 'email': email, 'department': department})
con.commit()
# Return the ID of the new employee
return jsonify({'id': cur.lastrowid}), 201
else:
return render_template('add_employee.html')
@app.route('/api/employees', methods=['POST'])
def add_employee():
# Extract the employee data from the request
data = request.get_json()
name = data['name']
email = data['email']
department = data['department']
# Check if the employee with the same ID already exists
cur = con.cursor()
cur.execute("SELECT * FROM employees WHERE email = :email", {'email': email})
employee = cur.fetchone()
if employee is not None:
return jsonify({'message': 'Employee with same email already exists'}), 400
# Insert the employee into the database
cur.execute("INSERT INTO employees (name, email, department) VALUES (:name, :email, :department)",
{'name': name, 'email': email, 'department': department})
con.commit()
@app.route('/api/employees/<int:id>', methods=['PUT'])
def update_employee(id):
# Check if the employee with the specified ID exists
cur = con.cursor()
cur.execute("SELECT * FROM employees WHERE id = :id", {'id': id})
employee = cur.fetchone()
if employee is None:
return jsonify({'message': 'Employee not found'}), 404
# Extract the employee data from the request
data = request.get_json()
name = data['name']
email = data['email']
department = data['department']
# Update the employee in the database
cur.execute("UPDATE employees SET name = :name, email = :email, department = :department WHERE id = :id",
{'name': name, 'email': email, 'department': department, 'id': id})
con.commit()
return jsonify({'message': 'Employee updated successfully'}), 200
@app.route('/api/employees/<int:id>', methods=['DELETE'])
def delete_employee(id):
# Check if the employee with the specified ID exists
cur = con.cursor()
cur.execute("SELECT * FROM employees WHERE id = :id", {'id': id})
employee = cur.fetchone()
if employee is None:
return jsonify({'message': 'Employee not found'}), 404
# Delete the employee from the database
cur.execute("DELETE FROM employees WHERE id = :id", {'id': id})
con.commit()
return jsonify({'message': 'Employee deleted successfully'}), 200
if __name__ == '__main__':
# Start the HTTPS server
app.run(host='0.0.0.0', port=4443, ssl_context=('/home/opc/ssl-certs/cert.pem', '/home/opc/ssl-certs/key.pem'))Create a HTML file in location defined in
app = Flask(__name__, template_folder='/home/opc')This will send the POST request to the “/api/add_employee” endpoint using function add_employee_form(). Save this file with name ’add_employee.html’
<form action="/api/add_employee" method="post">
<div class="form-group">
<label for="name">Name:</label>
<input type="text" class="form-control" id="name" name="name" required>
</div>
<div class="form-group">
<label for="email">Email:</label>
<input type="email" class="form-control" id="email" name="email" required>
</div>
<div class="form-group">
<label for="department">Department:</label>
<input type="text" class="form-control" id="department" name="department" required>
</div>
<button type="submit" class="btn btn-primary">Add Employee</button>
</form>5. Execute the Python script and Test the API’s using curl
$ python3 ~/oracle_flask_v3.py
* Running on https://10.180.1.21:4443/ (Press CTRL+C to quit)
# Get report of all employees
curl https://10.180.1.21:4443/api/employees -k
# Add a new employee
curl -X POST https://10.180.1.21:4443/api/employees -d '{"name":"Shadab M","email":"shadabm@example.com","department":"IT"}' -H "Content-Type: application/json" -k
# Get employee details of a particular employee using id
curl https://10.180.1.21:4443/api/employees/1 -k
# Update employee details using id
curl -X PUT https://10.180.1.21:4443/api/employees/1 -d '{"name":"Michael","email":"michael@example.com","department":"IT"}' -H "Content-Type: application/json" -k
curl https://10.180.1.21:4443/api/employees/1 -k
# Delete employee using id
curl -X DELETE https://10.180.1.21:4443/api/employees/1 -k
curl https://10.180.1.21:4443/api/employees/1 -k
# Search employee by email
curl https://10.180.1.21:4443/api/employees/search/shadabm@example.com -k6. Test Add Employee API using a HTML form
$ links https://10.180.1.21:4443/api/add_employee
If the log of the server returns HTTP1.1/ 201 this indicates the POST call via the html form was successful.
10.180.1.21 - - [15/Jan/2023 08:03:26] "POST /api/add_employee HTTP/1.1" 201 -
Conclusion
As we can easy, it is quite easy to build CRUD RESTful API’s on top of Oracle Database using Python3 and Flask web framework.