Python Database Access

Content:

1. Introduction
2. DB API 2.0
3. DB API 2.0 / Jython
4. SQLModel
5. NoSQL

Introduction:

Database access is important in all languages including Python.

I was not really expecting to write about Python database access as I don't know much about it, but since I have written about database access in so many other languages then I decided to write a little anyway.

DB API 2.0:

Python provide a standard for database access.

DB API 1.0 was defined in PEP 248 in 1996.

DB API 2.0 was defined in PEP 249 in 2001.

This will cover only DB API 2.0 (aka PEP 249).

DB API 2.0 is a very easy to use API (in the true Python spirit).

Note that DB API 2.0 is a relative flexiable/relaxed standard. There are some behavior left to implementation and some functionality is optional.

The implementation specific include:

• arguments to connect method
• parameter placeholders

Example:

import pymysql
import psycopg2
import cx_Oracle
import pymssql
import pyodbc

def t1_get_one(con, f2, pnam):
    c = con.cursor()
    c.execute('SELECT f1 FROM t1 WHERE f2 = ' + pnam(1), (f2,))
    f1 = c.fetchone()[0]
    c.close()
    return f1

def t1_get_all(con):
    c = con.cursor()
    c.execute('SELECT f1,f2 FROM t1')
    res = c.fetchall()
    c.close()
    return res

def t1_put(con, f1, f2, pnam):
    c = con.cursor()
    c.execute('INSERT INTO t1(f1,f2) VALUES(' + pnam(1) + ',' + pnam(2) + ')', (f1, f2))
    c.close()

def t1_remove(con, f1, pnam):
    c = con.cursor()
    c.execute('DELETE FROM t1 WHERE f1 = ' + pnam(1), (f1,))
    c.close()

def t1_display(data):     
    for row in data:
        print('%d %s' % (row[0], row[1]))

def test(lbl, con, pnam):
    print(lbl + ':')
    f1 = t1_get_one(con, 'BB', pnam)
    print(f1)
    data = t1_get_all(con)
    t1_display(data)
    t1_put(con, 999, 'XXX', pnam)
    data = t1_get_all(con)
    t1_display(data)
    t1_remove(con, 999, pnam)
    data = t1_get_all(con)
    t1_display(data)
    con.commit()
    con.close()

test('MySQL', pymysql.connect(host='localhost',user='root',password='',db='Test'), lambda n : '%s')
test('PostgreSQL', psycopg2.connect('host=localhost port=5432 dbname=Test user=postgres password=xxxxxx'), lambda n : '%s')
test('Oracle', cx_Oracle.connect('arne/xxxxxx@localhost:1521/xe'), lambda n : ':p' + str(n))
test('SQLServer', pymssql.connect(server='ARNEPC4\SQLEXPRESS', database='Test'), lambda n: '%s')
test('ODBC DSN SQLserver', pyodbc.connect('DSN=ARNEPC4_SQLSRV;UID=;PWD='), lambda n: '?')
test('ODBC DSN MySQL', pyodbc.connect('DSN=ARNEPC4_MYSQL;UID=root;PWD='), lambda n: '?')
test('ODBC DSN DB2', pyodbc.connect('DSN=ARNEPC4_DB2;UID=arne;PWD=xxxxxx'), lambda n: '?')
test('ODBC DSNless SQLserver', pyodbc.connect('Driver={SQL Server Native Client 11.0};Server=localhost;Database=Test;Trusted_Connection=yes;'), lambda n: '?')
test('ODBC DSNless MySQL', pyodbc.connect('Driver={MySQL ODBC 5.1 Driver};Server=localhost;Database=Test;User=root;Password=;'), lambda n: '?')
test('ODBC DSNless DB2', pyodbc.connect('Driver={IBM DB2 ODBC DRIVER};Hostname=localhost;Database=Test;Uid=arne;Pwd=xxxxxx;'), lambda n: '?')

The pnam function is used to handle the different parameter placeholders. Each implementation module does have a global variable paramstyle indicating what it expect, but that does not help when having multiple modules imported.

Same example using with:

import pymysql
import psycopg2
import cx_Oracle
import pymssql
import pyodbc

def t1_get_one(con, f2, pnam):
    with con.cursor() as c:
        c.execute('SELECT f1 FROM t1 WHERE f2 = ' + pnam(1), (f2,))
        f1 = c.fetchone()[0]
    return f1

def t1_get_all(con):
    with con.cursor() as c:
        c.execute('SELECT f1,f2 FROM t1')
        res = c.fetchall()
    return res

def t1_put(con, f1, f2, pnam):
    with con.cursor() as c:
        c.execute('INSERT INTO t1(f1,f2) VALUES(' + pnam(1) + ',' + pnam(2) + ')', (f1, f2))

def t1_remove(con, f1, pnam):
    with con.cursor() as c:
        c.execute('DELETE FROM t1 WHERE f1 = ' + pnam(1), (f1,))

def t1_display(data):
    for row in data:
        print('%d %s' % (row[0], row[1]))

def test(lbl, con, pnam):
    print(lbl + ':')
    f1 = t1_get_one(con, 'BB', pnam)
    print(f1)           
    data = t1_get_all(con)
    t1_display(data)
    t1_put(con, 999, 'XXX', pnam)
    data = t1_get_all(con)
    t1_display(data)
    t1_remove(con, 999, pnam)
    data = t1_get_all(con)
    t1_display(data)

cparam = lambda n : '%s'
qparam = lambda n : '?'
nparam = lambda n : ':p' + str(n)

con = pymysql.connect(host='localhost',user='root',password='',db='Test') # pymysql with return a cursor nor a connection
test('MySQL', con, cparam)
con.close()
with psycopg2.connect('host=localhost port=5432 dbname=Test user=postgres password=xxxxxx') as con:
    test('PostgreSQL', con, cparam)
with cx_Oracle.connect('arne/xxxxxx@localhost:1521/xe') as con:
    test('Oracle', con, nparam)
with pymssql.connect(server='ARNEPC4\SQLEXPRESS', database='Test') as con:
    test('SQLServer', con, cparam)
with pyodbc.connect('DSN=ARNEPC4_SQLSRV;UID=;PWD=') as con:
    test('ODBC DSN SQLserver', con, qparam)
with pyodbc.connect('DSN=ARNEPC4_MYSQL;UID=root;PWD=') as con:
    test('ODBC DSN MySQL', con, qparam)
with pyodbc.connect('DSN=ARNEPC4_DB2;UID=arne;PWD=xxxxxx') as con:
    test('ODBC DSN DB2', con, qparam)
with pyodbc.connect('Driver={SQL Server Native Client 11.0};Server=localhost;Database=Test;Trusted_Connection=yes;') as con:
    test('ODBC DSNless SQLserver', con, qparam)
with pyodbc.connect('Driver={MySQL ODBC 5.1 Driver};Server=localhost;Database=Test;User=root;Password=;') as con:
    test('ODBC DSNless MySQL', con, qparam)
with pyodbc.connect('Driver={IBM DB2 ODBC DRIVER};Hostname=localhost;Database=Test;Uid=arne;Pwd=xxxxxx;') as con:
    test('ODBC DSNless DB2', con, qparam)

DB API 2.0 / Jython:

Jython is Python implementation for JVM.

JDBC is standard database API for Java and other JVM languages.

An open source library zxJDBC translates from Python DB API 2.0 to JDBC.

By using zxJDBC Python code using DB API 2.0 can run using Jython.

Same examples as above:

from com.ziclix.python.sql import zxJDBC

def t1_get_one(con, f2, pnam):
    c = con.cursor()
    c.execute('SELECT f1 FROM t1 WHERE f2 = ' + pnam(1), (f2,))
    f1 = c.fetchone()[0]
    c.close()
    return f1

def t1_get_all(con):
    c = con.cursor()
    c.execute('SELECT f1,f2 FROM t1')
    res = c.fetchall()
    c.close()
    return res

def t1_put(con, f1, f2, pnam):
    c = con.cursor()
    c.execute('INSERT INTO t1(f1,f2) VALUES(' + pnam(1) + ',' + pnam(2) + ')', (f1, f2))
    c.close()

def t1_remove(con, f1, pnam):
    c = con.cursor()
    c.execute('DELETE FROM t1 WHERE f1 = ' + pnam(1), (f1,))
    c.close()

def t1_display(data):     
    for row in data:
        print('%d %s' % (row[0], row[1]))

def test(lbl, con, pnam):
    print(lbl + ':')
    f1 = t1_get_one(con, 'BB', pnam)
    print(f1)
    data = t1_get_all(con)
    t1_display(data)
    t1_put(con, 999, 'XXX', pnam)
    data = t1_get_all(con)
    t1_display(data)
    t1_remove(con, 999, pnam)
    data = t1_get_all(con)
    t1_display(data)
    con.commit()
    con.close()

test('MySQL', zxJDBC.connect('jdbc:mysql://localhost:3306/test', 'root', '', 'com.mysql.jdbc.Driver'), lambda n: '?')
test('DB2', zxJDBC.connect('jdbc:db2://localhost:50000/test', 'arne', 'xxxxxx', 'com.ibm.db2.jcc.DB2Driver'), lambda n: '?')
test('Oracle', zxJDBC.connect('jdbc:oracle:thin:@localhost:1521:xe', 'arne', 'xxxxxx', 'oracle.jdbc.OracleDriver'), lambda n: '?')

from com.ziclix.python.sql import zxJDBC

def t1_get_one(con, f2, pnam):
    with con.cursor() as c:
        c.execute('SELECT f1 FROM t1 WHERE f2 = ' + pnam(1), (f2,))
        f1 = c.fetchone()[0]
    return f1

def t1_get_all(con):
    with con.cursor() as c:
        c.execute('SELECT f1,f2 FROM t1')
        res = c.fetchall()
    return res

def t1_put(con, f1, f2, pnam):
    with con.cursor() as c:
        c.execute('INSERT INTO t1(f1,f2) VALUES(' + pnam(1) + ',' + pnam(2) + ')', (f1, f2))

def t1_remove(con, f1, pnam):
    with con.cursor() as c:
        c.execute('DELETE FROM t1 WHERE f1 = ' + pnam(1), (f1,))

def t1_display(data):
    for row in data:
        print('%d %s' % (row[0], row[1]))

def test(lbl, con, pnam):
    print(lbl + ':')
    f1 = t1_get_one(con, 'BB', pnam)
    print(f1)           
    data = t1_get_all(con)
    t1_display(data)
    t1_put(con, 999, 'XXX', pnam)
    data = t1_get_all(con)
    t1_display(data)
    t1_remove(con, 999, pnam)
    data = t1_get_all(con)
    t1_display(data)

qparam = lambda n : '?'

with zxJDBC.connect('jdbc:mysql://localhost:3306/test', 'root', '', 'com.mysql.jdbc.Driver') as con:
    test('MySQL', con, qparam)
with zxJDBC.connect('jdbc:db2://localhost:50000/test', 'arne', 'xxxxxx', 'com.ibm.db2.jcc.DB2Driver') as con:
    test('DB2', con, qparam)
with zxJDBC.connect('jdbc:oracle:thin:@localhost:1521:xe', 'arne', 'xxxxxx', 'oracle.jdbc.OracleDriver') as con:
    test('Oracle', con, qparam)

Jython obviously needs both zxJDBC and the JDBC driver in classpath.

SQLModel:

I am generally not impressed by ORM frameworks for Python, but a new project SQLModel looks very promising.

SQLModel is an open source ORM framework for Python.

It is available from Github and can be installed with pip install sqlmodel.

It requires Python 3.6 or newer.

It is brand new. First release was in August 2021 and at the time of writing the version is 0.0.4.

Example:

from sqlmodel import Field, Session, SQLModel, create_engine, select

class T1(SQLModel, table=True):
    f1: int = Field(default=None, primary_key=True)
    f2: str

def t1_get_one(eng, f2):
    with Session(eng) as ses:
        selone = select(T1).where(T1.f2 == f2)
        res = ses.exec(selone)
        f1 = res.first().f1
    return f1

def t1_get_all(eng):
    with Session(eng) as ses:
        selall = select(T1)
        res = ses.exec(selall)
        reslst = res.all()
    return reslst

def t1_put(eng, f1, f2):
    with Session(eng) as ses:
        o = T1(f1=f1, f2=f2);
        ses.add(o)
        ses.commit()

def t1_remove(eng, f1):
    with Session(eng) as ses:
        selone = select(T1).where(T1.f1 == f1)
        o = ses.exec(selone).first()
        ses.delete(o)
        ses.commit()

def t1_display(lst):     
    for o in lst:
        print('%d %s' % (o.f1, o.f2))

def test(lbl, eng):
    print(lbl + ':')
    f1 = t1_get_one(eng, 'BB')
    print(f1)
    lst = t1_get_all(eng)
    t1_display(lst)
    t1_put(eng, 999, 'XXX')
    lst = t1_get_all(eng)
    t1_display(lst)
    t1_remove(eng, 999)
    lst = t1_get_all(eng)
    t1_display(lst)

test('MySQL', create_engine('mysql+pymysql://root:@localhost/Test'))
test('PostgreSQL', create_engine('postgresql+psycopg2://postgres:xxxxxx@localhost/Test'))
test('ODBC DSN MySQL', create_engine('mysql+pyodbc://:@ARNEPC4_MYSQL'))
##test('ODBC DSN PostgreSQL', create_engine('postgresql+pyodbc://:@ARNEPC4_PGSQL'))
##test('SQLServer', create_engine('mssql+pymssql://:@localhost/Test'))
##test('ODBC DSN SQLServer', create_engine('mssql+pyodbc://:@ARNEPC4_SQLSRV'))

It does not work for all databases yet.

NoSQL:

All of the above are for relational (SQL) databases. For NoSQL databases see:

• DBM
• MongoDB
• OrientDB

Article history:

Other articles:

See list of all articles here

Comments:

Please send comments to Arne Vajhøj