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 put 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 concurrent
    # 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 a new 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')))

Join the discussion

Comments ( 6 )
  • Leonel Fonseca Thursday, December 3, 2020
    Nice example!
  • Jorge Perez Monday, December 7, 2020
    Excellent guide. Thanks.
  • Sidney Kim Wednesday, February 17, 2021
    This is cool! Thanks a lot!
  • Vicente Coopman Friday, February 26, 2021
    Nice guide! But i having a problem i don't know how to fix. It is about the connections in the pool seems to get caducated after some time. When i deploy the application, all seems to work correctly but after a while, the cursor gets stuck when it trys to execute and statement.
    This is my Code:
    import sys
    import logging
    import cx_Oracle

    logger = logging.getLogger(__name__)

    def init_pool(app):
    ''' adds pool of database connections to app '''

    try:
    # Get credentials and configs values
    db_user = app.config['DB_USER']
    db_pass = app.config['DB_PASS']
    db_url = app.config['DB_URL']

    except ValueError as ex:
    logger.error("Missing database config value %s in current app", ex)
    logger.error(ex)
    sys.exit()

    pool_min = 4
    pool_max = 4
    pool_incr = 0 # Fixed size pool
    pool_gmd = cx_Oracle.SPOOL_ATTRVAL_WAIT

    try:
    # Create pool
    logger.info("Creating pool of conections...")
    pool = cx_Oracle.SessionPool(
    user=db_user,
    password=db_pass,
    dsn=db_url,
    min=pool_min,
    max=pool_max,
    increment=pool_incr,
    threaded=True,
    getmode=pool_gmd)

    logger.info("Pool Created")
    app.config['pool'] = pool
    logger.info("Pool added to app")

    except Exception as ex:
    logger.error("Unnable to create pool of connections to database")
    logger.error(ex)
    sys.exit()

    any help will be really appreciated!
  • Christopher Jones Friday, February 26, 2021
    Blog comments aren't a great place to debug issues - particularly if only limited information is shared. A common problem with a 'stuck cursor' could be that another user or app has locked some table rows by forgetting to commit.
  • Philipp Abele Thursday, May 6, 2021
    Thank you Christpher for this great guide and the provided source.

    However, I am using connexion to have the benefits of a swagger ui.

    Unfortunately, I don't know how to share the pool here. I would appreciate to get in contact to solve this issue.
Please enter your name.Please provide a valid email address.Please enter a comment.CAPTCHA challenge response provided was incorrect. Please try again.