Building a POST REST API with Flask and Oracle Database

 In my previous articles in the Oracle + Flask series, we explored how to connect Flask with Oracle Database and read data from an Oracle table.

We then took the next step by exposing Oracle data through a GET REST API.

Now let's take it one step further.

In this tutorial, we will create a POST REST API that accepts employee information in JSON format and inserts that data into our Oracle table.

The flow will look like this:

JSON Request → Flask REST API → SQLAlchemy → Oracle Database

This demonstrates the complete integration:

             POST Request
                  │
                  ▼
        ┌─────────────────┐
        │   Flask API     │
        │ /EmployeeInsert │
        └────────┬────────┘
                 │
                 ▼
          JSON Validation
                 │
                 ▼
          SQLAlchemy ORM
                 │
                 ▼
          Oracle Database
                 │
                 ▼
          JSON Response

This is a very common pattern when building integrations between applications and Oracle Database.

What Are We Building?

We already have our Oracle table:

XXFLASK_EMPLOYEE_INFO

Now we'll create the following REST endpoint:

POST /EmployeeInsert

The API will accept JSON like this:

{
    "name": "Pranay",
    "dept": "IT",
    "salary": 60000,
    "contact": "+9195555552367"
}

The Flask application will receive this JSON, validate the required fields, create an employee record, and insert it into Oracle Database.

If everything is successful, the API will return:

{
    "message": "Employee Successfully Created",
    "name": "Pranay",
    "dept": "IT"
}

---

1. Import the Required Libraries

Let's start with the imports.

from flask import Flask, render_template, request, jsonify
from flask_sqlalchemy import SQLAlchemy
import oracledb
import json
from datetime import datetime
import os

The important libraries for this API are:

Library	Purpose
Flask	Creates our web application
request	Reads data sent by the client
jsonify	Returns JSON responses
Flask-SQLAlchemy	Communicates with the database through ORM
oracledb	Oracle Database connectivity
json	Reads configuration
datetime	Stores creation date/time
os	Handles configuration file paths

For this particular API, request and jsonify are especially important.

---

2. Read the Database Configuration

Just like in our previous application, we don't want to hardcode the database connection directly into our Python code.

We'll continue using our json.config file.

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

config_path = os.path.join(BASE_DIR, 'json.config')

with open(config_path, 'r') as c:
    DB_PARAMS = json.load(c)['params']

Our configuration file contains the database connection:

{
    "params": {
        "database_conn_link": "oracle+oracledb://apps:apps@orcl:1522/?service_name=devdb"
    }
}

This approach makes the application configurable.

If we need to move the application from Development to Test or Production, we can change the configuration rather than modifying the application code.

---

3. Initialize Oracle and Flask

Next, initialize the Oracle Client.

oracledb.init_oracle_client(
    lib_dir=r"C:\instantclient_21_11"
)

Then create the Flask application:

app = Flask(__name__)

Configure SQLAlchemy:

app.config['SQLALCHEMY_DATABASE_URI'] = \
    DB_PARAMS['database_conn_link']

db = SQLAlchemy(app)

At this point, Flask and SQLAlchemy are configured to communicate with our Oracle Database.

---

4. Create the Employee Model

Our Oracle table is:

XXFLASK_EMPLOYEE_INFO

We represent this table using a SQLAlchemy model.

class XXFlaskEmployeeInfo(db.Model):

    __tablename__ = 'XXFLASK_EMPLOYEE_INFO'

    XXID = db.Column("XXID", db.Integer, primary_key=True)

    NAME = db.Column("NAME", db.String)

    DEPT = db.Column("DEPT", db.String)

    SALARY = db.Column("SALARY", db.Integer)

    CONTACT_INFO = db.Column("CONTACT_INFO", db.String)

    CREATION_DATE = db.Column("CREATION_DATE", db.Date)

The important concept here is that the Python class represents the Oracle table.

For example:

Python                     Oracle
------------------------------------------------
XXFlaskEmployeeInfo   →    XXFLASK_EMPLOYEE_INFO
NAME                  →    NAME
DEPT                  →    DEPT
SALARY                →    SALARY
CONTACT_INFO          →    CONTACT_INFO

SQLAlchemy handles the interaction between our Python objects and the Oracle table.

---

5. Create the POST REST Endpoint

Now comes the most important part of this tutorial.

@app.route('/EmployeeInsert', methods=['POST'])

This tells Flask that /EmployeeInsert is our REST endpoint and that it accepts the POST HTTP method.

The complete function is:

def postEmployeeInfo():

    message = None

    if request.method == 'POST':

        try:

            data = request.get_json()

            if not data.get("name"):
                return jsonify({
                    "error": "Name is required"
                }), 400

            if not data.get("dept"):
                return jsonify({
                    "error": "Dept is required"
                })

            if not data.get("salary"):
                return jsonify({
                    "error": "Salary is required"
                }), 400

            if not data.get("contact"):
                return jsonify({
                    "error": "Contacts are essentials"
                }), 400

            EmployeePost = XXFlaskEmployeeInfo(
                NAME=data['name'],
                DEPT=data["dept"],
                SALARY=data["salary"],
                CONTACT_INFO=data["contact"],
                CREATION_DATE=datetime.utcnow()
            )

            db.session.add(EmployeePost)

            db.session.commit()

            lSuccessMessage = {
                "message": "Employee Successfully Created",
                "name": EmployeePost.NAME,
                "dept": EmployeePost.DEPT
            }

            return jsonify(lSuccessMessage)

        except Exception as e:

            return jsonify({
                "error": str(e)
            }), 500

Let's break this down.

---

6. Reading JSON from the Request

The client sends employee information in the request body.

We retrieve that JSON using:

data = request.get_json()

For example, the client sends:

{
    "name": "Rahul",
    "dept": "IT",
    "salary": 55000,
    "contact": "+9195555551111"
}

After executing:

data = request.get_json()

we can access the values like:

data['name']
data['dept']
data['salary']
data['contact']

This is one of the most important concepts when building REST APIs with Flask.

---

7. Validate the Input

Before inserting data into Oracle, we should validate the request.

For example:

if not data.get("name"):
    return jsonify({
        "error": "Name is required"
    }), 400

If the client doesn't provide a name, the API returns:

{
    "error": "Name is required"
}

The HTTP status code is:

400 Bad Request

We perform similar validation for:

• Name
• Department
• Salary
• Contact

This prevents incomplete data from being inserted into the database.

---

8. Create the Database Object

Once validation succeeds, we create a Python object representing the employee record.

EmployeePost = XXFlaskEmployeeInfo(
    NAME=data['name'],
    DEPT=data["dept"],
    SALARY=data["salary"],
    CONTACT_INFO=data["contact"],
    CREATION_DATE=datetime.utcnow()
)

Think of this as creating a new record in memory.

We haven't inserted it into Oracle yet.

---

9. Insert the Record into Oracle

Now we add the object to the SQLAlchemy session.

db.session.add(EmployeePost)

Then commit the transaction:

db.session.commit()

The commit() is important.

It makes the transaction permanent in the database.

Conceptually, our application is doing:

JSON
  ↓
Python Object
  ↓
SQLAlchemy Session
  ↓
INSERT
  ↓
Oracle Database

---

10. Return a Success Response

After the database commit succeeds, we create a response.

lSuccessMessage = {
    "message": "Employee Successfully Created",
    "name": EmployeePost.NAME,
    "dept": EmployeePost.DEPT
}

Then:

return jsonify(lSuccessMessage)

The API returns:

{
    "message": "Employee Successfully Created",
    "name": "Rahul",
    "dept": "IT"
}

This gives the calling application confirmation that the employee was successfully created.

---

11. Exception Handling

Database operations can fail for many reasons.

For example:

• Database unavailable
• Invalid database connection
• Constraint violation
• Invalid data type
• SQLAlchemy error

That's why the code uses:

try:
    ...
except Exception as e:
    return jsonify({
        "error": str(e)
    }), 500

If something goes wrong, the API returns an HTTP 500 response.

Example:

{
    "error": "Database connection error..."
}

12. Test the API Using Postman

Now let's test our API.

Select:

Body → raw → JSON

Then provide:

{
    "name": "Rahul",
    "dept": "IT",
    "salary": 55000,
    "contact": "+9195555551111"
}

Click Send.

The API should return:

{
    "message": "Employee Successfully Created",
    "name": "Rahul",
    "dept": "IT"
}


13. Verify the Data in Oracle
Now we can verify that the record was inserted into Oracle.
SELECT
    XXID,
    NAME,
    DEPT,
    SALARY,
    CONTACT_INFO,
    CREATION_DATE
FROM XXFLASK_EMPLOYEE_INFO
ORDER BY XXID;

You should see the newly created employee record.


Complete Application
Here is the complete code together:

from flask import Flask, render_template,request,jsonify
from flask_sqlalchemy import SQLAlchemy
import oracledb
import json
from datetime import datetime
import os
BASE_DIR=os.path.dirname(os.path.abspath(__file__))
config_path=os.path.join(BASE_DIR,'json.config')
with open(config_path,'r') as c:
 DB_PARAMS= json.load(c)['params'] #Database Parms
oracledb.init_oracle_client(lib_dir=r"C:\instantclient_21_11")
app=Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI']=DB_PARAMS['database_conn_link']
db=SQLAlchemy(app)
class XXFlaskEmployeeInfo(db.Model):
  __tablename__='XXFLASK_EMPLOYEE_INFO'
  XXID=db.Column("XXID",db.Integer,primary_key=True)
  NAME=db.Column("NAME",db.String)
  DEPT=db.Column("DEPT",db.String)
  SALARY=db.Column("SALARY",db.Integer)
  CONTACT_INFO=db.Column("CONTACT_INFO",db.String)
  CREATION_DATE=db.Column("CREATION_DATE",db.Date)
@app.route('/EmployeeInsert',  methods=['POST'])
def postEmployeeInfo():
   message = None
   if request.method == 'POST':
     try:
       data=request.get_json() #get json from input body params
       if not data.get("name"):
         return jsonify({"error":"Name is required"}),400
       if not data.get("dept"):
         return jsonify({"error":"Dept is required"})
       if not data.get("salary"):
         return jsonify({"error":"Salary is required"}),400
       if not data.get("contact"):
         return jsonify({"error":"Contacts are essentials"}),400
       EmployeePost= XXFlaskEmployeeInfo(NAME=data['name'],DEPT=data["dept"],SALARY=data["salary"],CONTACT_INFO=data["contact"],CREATION_DATE=datetime.utcnow())
       db.session.add(EmployeePost)
       db.session.commit()
       lSuccessMessage={"message":"Employee Successfully Created","name":EmployeePost.NAME,"dept":EmployeePost.DEPT}
       return jsonify (lSuccessMessage)
     except Exception as e:
      return jsonify({"error": str(e)}), 500
if __name__=="__main__":
   app.run(debug=True)

---