X

The leading edge of scripting languages and Oracle Database brought to you by the Data Access Development team

How to use Python Flask with Oracle Database

Christopher Jones
Senior Principal Product Manager

cx_Oracle logo

This post shows how to install and create a Python Flask application for Oracle Database.

Flask is a Python web framework. It is a 'micro framework' that doesn't enforce dependencies, so Oracle's excellent cx_Oracle module can be easily used to work with the database.

  

The demo app is a little service that lets you insert people's names into a 'user' table. You can query names by their id number.

Flask apps typically handle multiple parallel application users, so it's important to create a pool of database connections in the application. Other sample apps tends to skip this, so I particularly wanted to show it. Connection pooling is important for performance when applications frequently connect and disconnect from the database. Not only are pools faster, but they also provide and support Oracle's high availability features. They are recommended for all applications that must be reliable. This also means that small pools can be useful for applications that want a few connections available for infrequent use.

To run the application:

  • Install Python 3, if not already installed.

  • Install the Flask module: pip install --upgrade Flask. For details see the Flask installation documentation.

  • Install Oracle's cx_Oracle 8 module and the Oracle Client libraries (if they not already available): pip install --upgrade cx_Oracle. For details see the Quick Start cx_Oracle Installation documentation.

    You can get Oracle Instant Client freely from the Oracle Instant Client Downloads page. Note you don't need this if Python is running on the same computer as your database.

  • If you don't have a database, you can install Oracle Express Edition or create a permanent, free Oracle Cloud Database. Mac users can check out The Easiest Way to Install Oracle Database on Apple macOS.

  • Save the code shown below as demo.py. (You can also download it from demo.zip.)

    Review demo.py. Note I've used cx_Oracle 8's init_oracle_client() function at line 28 to tell cx_Oracle where to find the Oracle Client libraries. Check the cx_Oracle 8 Initialization chapter for details on how this is used, or whether it is needed in your environment.

    In the demo, a pool of connections is created in start_pool() (at line 70) during application startup. Then each application route will get a connection from the pool and release it back to the pool after using it.

    The main Flask functionality is provided by three routes.

    • The first for / is implemented at line 141 in index() and simply returns a welcome message

    • A /post/<username> route is handled at line 150 by post() which inserts the given name.

    • A /users/<n> route is handled at line 163 by show_username() which queries the database and returns the username associated with given id number.

  • Set the environment variables PYTHON_USERNAME and PYTHON_PASSWORD to your database username and password. Set PYTHON_CONNECTSTRING to the connection string for your database. For example if you have installed Oracle XE then you would set it to something like "example.com/XEPDB1".

    $ export PYTHON_USERNAME=cj
    $ export PYTHON_PASSWORD=welcome
    $ export PYTHON_CONNECTSTRING=example.com/XEPDB1
    
  • Run the file. For simple testing I run:

    $ python demo.py
    

    but you may prefer a more official Flask invocation, for example FLASK_APP=demo.py flask run. See the Flask quickstart.

    Output will be like:

    $ python demo.py 
    Connecting to example.com/orclpdb1
     * Serving Flask app "demo" (lazy loading)
     * Environment: production
       WARNING: This is a development server. Do not use it in a production deployment.
       Use a production WSGI server instead.
     * Debug mode: off
     * Running on http://127.0.0.1:8080/ (Press CTRL+C to quit)
    
  • In a browser, open the URL http://127.0.0.1:8080/ and you will see the demo message from the index() function:

  • Now query the precreated user who has an id of 1. Load http://127.0.0.1:8080/user/1

  • If you try to query an id that doesn't exist with http://127.0.0.1:8080/user/2 you'll get the error message:

  • To insert a new name, load http://127.0.0.1:8080/post/alison

  • Now this name can be queried back

That's it from the Oracle side. The rest is all beautification.

Demo Code

The demo application code is below (download from demo.zip):

"""
demo.py

Christopher Jones, 10 Sep 2020

Demo of using flask with Oracle Database

Before running, set these environment variables:

    PYTHON_USERNAME       - your DB username
    PYTHON_PASSWORD       - your DB password
    PYTHON_CONNECTSTRING  - the connection string to the DB, e.g. "example.com/XEPDB1"
    PORT                  - port the web server will listen on.  The default in 8080

"""

import os
import sys
import cx_Oracle
from flask import Flask

################################################################################
#
# On macOS tell cx_Oracle 8 where the Instant Client libraries are.  You can do
# the same on Windows, or add the directories to PATH.  On Linux, use ldconfig
# or LD_LIBRARY_PATH.  cx_Oracle installation instructions are at:
# https://cx-oracle.readthedocs.io/en/latest/user_guide/installation.html
if sys.platform.startswith("darwin"):
    cx_Oracle.init_oracle_client(lib_dir=os.environ.get("HOME")+"/instantclient_19_3")
elif sys.platform.startswith("win32"):
    cx_Oracle.init_oracle_client(lib_dir=r"c:\oracle\instantclient_19_8")

################################################################################
#
# Start a connection pool.
#
# Connection pools allow multiple, concurrent web requests to be efficiently
# handled.  The alternative would be to open a new connection for each use
# which would be very slow, inefficient, and not scalable.  Connection pools
# support Oracle high availability features.
#
# Doc link: https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html#connection-pooling


# init_session(): a 'session callback' to efficiently set any initial state
# that each connection should have.
#
# If you have multiple SQL statements, then them all in a PL/SQL anonymous
# block with BEGIN/END so you only call execute() once.  This is shown later in
# create_schema().
#
# This particular demo doesn't use dates, so sessionCallback could be omitted,
# but it does show settings many apps would use.
#
# Note there is no explicit 'close cursor' or 'close connection'.  At the
# end-of-scope when init_session() finishes, the cursor and connection will be
# closed automatically.  In real apps with a bigger code base, you will want to
# close each connection as early as possible so another web request can use it.
#
# Doc link: https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html#session-callbacks-for-setting-pooled-connection-state
#
def init_session(connection, requestedTag_ignored):
    cursor = connection.cursor()
    cursor.execute("""
        ALTER SESSION SET
          TIME_ZONE = 'UTC'
          NLS_DATE_FORMAT = 'YYYY-MM-DD HH24:MI'""")

# start_pool(): starts the connection pool
def start_pool():

    # Generally a fixed-size pool is recommended, i.e. pool_min=pool_max.
    # Here the pool contains 4 connections, which is fine for 4 conncurrent
    # users.
    #
    # The "get mode" is chosen so that if all connections are already in use, any
    # subsequent acquire() will wait for one to become available.

    pool_min = 4
    pool_max = 4
    pool_inc = 0
    pool_gmd = cx_Oracle.SPOOL_ATTRVAL_WAIT

    print("Connecting to", os.environ.get("PYTHON_CONNECTSTRING"))

    pool = cx_Oracle.SessionPool(user=os.environ.get("PYTHON_USERNAME"),
                                 password=os.environ.get("PYTHON_PASSWORD"),
                                 dsn=os.environ.get("PYTHON_CONNECTSTRING"),
                                 min=pool_min,
                                 max=pool_max,
                                 increment=pool_inc,
                                 threaded=True,
                                 getmode=pool_gmd,
                                 sessionCallback=init_session)

    return pool

################################################################################
#
# create_schema(): drop and create the demo table, and add a row
#
def create_schema():
    connection = pool.acquire()
    cursor = connection.cursor()
    cursor.execute("""
        begin
          begin
            execute immediate 'drop table demo';
            exception when others then
              if sqlcode <> -942 then
                raise;
              end if;
          end;

          execute immediate 'create table demo (
               id        number generated by default as identity,
               username varchar2(40))';

          execute immediate 'insert into demo (username) values (''chris'')';

          commit;
        end;""")

################################################################################
#
# Specify some routes
#
# The default route will display a welcome message:
#   http://127.0.0.1:8080/
#
# To insert new a user 'fred' you can call:
#    http://127.0.0.1:8080/post/fred
#
# To find a username you can pass an id, for example 1:
#   http://127.0.0.1:8080/user/1
#

app = Flask(__name__)

# Display a welcome message on the 'home' page
@app.route('/')
def index():
    return "Welcome to the demo app"

# Add a new username
#
# The new user's id is generated by the DB and returned in the OUT bind
# variable 'idbv'.  As before, we leave closing the cursor and connection to
# the end-of-scope cleanup.
@app.route('/post/<string:username>')
def post(username):
    connection = pool.acquire()
    cursor = connection.cursor()
    connection.autocommit = True
    idbv = cursor.var(int)
    cursor.execute("""
        insert into demo (username)
        values (:unbv)
        returning id into :idbv""", [username, idbv])
    return 'Inserted {} with id {}'.format(username, idbv.getvalue()[0])

# Show the username for a given id
@app.route('/user/<int:id>')
def show_username(id):
    connection = pool.acquire()
    cursor = connection.cursor()
    cursor.execute("select username from demo where id = :idbv", [id])
    r = cursor.fetchone()
    return (r[0] if r else "Unknown user id")

################################################################################
#
# Initialization is done once at startup time
#
if __name__ == '__main__':

    # Start a pool of connections
    pool = start_pool()

    # Create a demo table
    create_schema()

    # Start a webserver
    app.run(port=int(os.environ.get('PORT', '8080')))

Be the first to comment

Comments ( 0 )
Please enter your name.Please provide a valid email address.Please enter a comment.CAPTCHA challenge response provided was incorrect. Please try again.