How to connect to various DBs from Python (PEP 249) and SQLAlchemy

1.First of all

Access to DB from python is specified as PEP 249 --Python Database API Specification v2.0, and each DB is implementing according to this specification. (This implementation provided by each DB will be called DBAPI.)

The idea is

 You can connect to the DB, execute SQL, and get the result with the same code without being aware of the DB of the connection destination.

I think that's what it means. So how is it actually? I would like to confirm.

2. About DB API prepared for each DB

The typical DB API of each DB is as shown in the list below.

Let's look at the typical DB API of each DB one by one.

2-1.MySQL

■ Implementation example using mysql module ("5.4 Querying Data Using Connector / Python" "Than)

import datetime
import mysql.connector

cnx = mysql.connector.connect(user='scott', database='employees')
cursor = cnx.cursor()

query = ("SELECT first_name, last_name, hire_date FROM employees "
         "WHERE hire_date BETWEEN %s AND %s")

hire_start = datetime.date(1999, 1, 1)
hire_end = datetime.date(1999, 12, 31)

cursor.execute(query, (hire_start, hire_end))

for (first_name, last_name, hire_date) in cursor:
  print("{}, {} was hired on {:%d %b %Y}".format(
    last_name, first_name, hire_date))

cursor.close()
cnx.close()

2-2.MariaDB

■ Implementation example using mariadb module ("How to connect Python programs to MariaDB" Than)

import mariadb 

conn = mariadb.connect(
    user="db_user",
    password="db_user_passwd",
    host="localhost",
    database="employees")
cur = conn.cursor() 

retrieving information 
some_name = "Georgi" 
cur.execute("SELECT first_name,last_name FROM employees WHERE first_name=?", (some_name,)) 

for first_name, last_name in cur: 
    print(f"First name: {first_name}, Last name: {last_name}")
    
insert information 
try: 
    cur.execute("INSERT INTO employees (first_name,last_name) VALUES (?, ?)", ("Maria","DB")) 
except mariadb.Error as e: 
    print(f"Error: {e}")

conn.commit() 
print(f"Last Inserted ID: {cur.lastrowid}")
    
conn.close()

2-3.PostgreSQL

■ Implementation example using psycopg2 module (from "Basic module usage")

>>> import psycopg2

 Connect to an existing database
>>> conn = psycopg2.connect("dbname=test user=postgres")

 Open a cursor to perform database operations
>>> cur = conn.cursor()

 Execute a command: this creates a new table
>>> cur.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);")

 Pass data to fill a query placeholders and let Psycopg perform
 the correct conversion (no more SQL injections!)
>>> cur.execute("INSERT INTO test (num, data) VALUES (%s, %s)",
...      (100, "abc'def"))

 Query the database and obtain data as Python objects
>>> cur.execute("SELECT * FROM test;")
>>> cur.fetchone()
(1, 100, "abc'def")

 Make the changes to the database persistent
>>> conn.commit()

 Close communication with the database
>>> cur.close()
>>> conn.close()

2-4.Oracle

■ Implementation example using psycopg2 module ("[Python and Oracle Database Tutorial: Scripting for the Future](https://oracle.github.io/python-cx_Oracle/samples/tutorial/Python-and-Oracle-Database-Scripting]" -for-the-Future.html) ")

import cx_Oracle
import db_config

con = cx_Oracle.connect(db_config.user, db_config.pw, db_config.dsn)

cur = con.cursor()
cur.execute("select * from dept order by deptno")
res = cur.fetchall()
for row in res:
    print(row)

cur.close()
con.close()

2-5.Microsoft SQL Server

■ Implementation example using the pymssql module ("[Step 3: Proof of concept for connecting to SQL using pymssql](https://docs.microsoft.com/ja-jp/sql/connect/python/pymssql/" step-3-proof-of-concept-connecting-to-sql-using-pymssql?view=sql-server-ver15) ")

import pymssql  
conn = pymssql.connect(server='yourserver.database.windows.net', user='yourusername@yourserver', password='yourpassword', database='AdventureWorks')  
cursor = conn.cursor()  
cursor.execute('SELECT c.CustomerID, c.CompanyName,COUNT(soh.SalesOrderID) AS OrderCount FROM SalesLT.Customer AS c LEFT OUTER JOIN SalesLT.SalesOrderHeader AS soh ON c.CustomerID = soh.CustomerID GROUP BY c.CustomerID, c.CompanyName ORDER BY OrderCount DESC;')  
row = cursor.fetchone()  
while row:  
    print str(row[0]) + " " + str(row[1]) + " " + str(row[2])     
    row = cursor.fetchone()  

2-6. Differences in DB API of each DB

If you look closely at the sample code above, you can see that in any DB API,

① Create a Connection instance ② Create a Cursor instance ③ Execute SQL using the generated Cursor instance ④ Fetch the execution result of SQL

I think it is in the form of. There are some differences in writing, but once you learn how to write one DBAPI, you can use another DB without any trouble. In this case, it seems good to use the DB API of each DB directly, but unfortunately the DB API standard does not include the Connection Pool specification. Since each DB API may have its own implementation, it can be incorporated into the application to be developed, but the implementation will be different depending on the DB used.

3.SQLAlchemy SQLAlchemy is known as ORM, but it is like an extension module of DBAPI that has various DBAPIs of various DBs and implements Connction Pool.

To be honest, using ORM is a hassle, and although it becomes a DB-dependent writing method, I always think that I want to write SQL directly. Even for such a selfish me, SQLAlchemy does not impose the use of ORM, and I am very grateful that it provides a means to execute SQL directly. If you are thinking of connecting to the DB with the DB API, please consider using SQLAlchemy.

Since it is not a DBAPI, the implementation method is slightly different, but you can operate the DB in almost the same flow as described in [2-6. Differences in DBAPI of each DB].

■ Implementation example using sqlalchemy module (code excerpt from "SQL Expression Language Tutorial")

>>> from sqlalchemy import create_engine
>>> engine = create_engine('sqlite:///:memory:', echo=True)
>>> conn = engine.connect()
>>> from sqlalchemy.sql import text
>>> s = text(
 ...     "SELECT users.fullname || ', ' || addresses.email_address AS title "
...         "FROM users, addresses "
...         "WHERE users.id = addresses.user_id "
...         "AND users.name BETWEEN :x AND :y "
...         "AND (addresses.email_address LIKE :e1 "
...             "OR addresses.email_address LIKE :e2)")
>>> conn.execute(s, x='m', y='z', e1='%@aol.com', e2='%@msn.com').fetchall()

If you want to use Connection Pool, describe the part of create_engine as follows.

■ From "Connection Pooling"

engine = create_engine('postgresql://me@localhost/mydb',pool_size=20, max_overflow=0)

4. Reference

PlaySQLAlchemy: Getting Started with SQLAlchemy The Novice’s Guide to the Python 3 DB-API Introduction to SQLAlchemy that seems to be usable for the time being (* ORM function is not used)