Manipulate various databases with Python
Describes how to operate the main database in Python. Python is running on both 2.x and 3.x in a Windows environment.
For test data, unzip T01Prefecture.zip on the following page and use the table and data.
__PHP programming beginner's introductory course __ http://php5.seesaa.net/article/61269550.html
MySQL
environment
Python 2.7 or 3.3 MySQL 5.6.15
Database settings
table
CREATE TABLE `t01prefecture` (
`PREF_CD` int(3) NOT NULL DEFAULT '0',
`PREF_NAME` varchar(10) DEFAULT NULL,
PRIMARY KEY (`PREF_CD`)
) ENGINE=InnoDB D
__ Stored __ that returns a single recordset
DELIMITER $$
CREATE DEFINER=`username`@`%` PROCEDURE `test_sp`(IN fromNo INT,
IN toNo INT)
BEGIN
select * from t01prefecture WHERE PREF_CD BETWEEN fromNo AND toNo ;
END$$
DELIMITER ;
__ Stored __ that returns multiple recordsets
DELIMITER $$
CREATE DEFINER=`username`@`%` PROCEDURE `test_sp2`(IN cd1 INT,IN cd2 INT)
BEGIN
select * from t01prefecture WHERE PREF_CD = cd1;
select * from t01prefecture WHERE PREF_CD = cd2;
END$$
DELIMITER ;
__Function example __
DELIMITER $$
CREATE DEFINER=`username`@`%` FUNCTION `test_fn`(cd INT) RETURNS varchar(10) CHARSET utf8
BEGIN
DECLARE ret VARCHAR(10);
SELECT PREF_NAME INTO ret FROM t01prefecture WHERE PREF_CD = cd;
RETURN ret;
END$$
DELIMITER ;
Python code
Use mysql-connector-python http://dev.mysql.com/downloads/connector/python/
# -*- coding: cp932 -*-
#Operation sample of mysql
#easy_install mysql-connector-python
import mysql.connector
try:
cnn = mysql.connector.connect(host='localhost',
port=3306,
db='Sample001',
user='root',
passwd='root',
charset="cp932")
cur = cnn.cursor()
#Organize test data
pref_cd = 100
cur.execute("""DELETE FROM t01prefecture WHERE PREF_CD >= %s""" , (pref_cd,))
cnn.commit()
print("Simple SELECT statement==========================")
from_id = 45
to_id = 999
#The following should match the character code of the environment!
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN %s AND %s""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Commit test==========================")
pref_cd = 100
pref_name = "Motemote Kingdom"
cur.execute("""INSERT INTO t01prefecture(PREF_CD, PREF_NAME)
VALUES (%s, %s)""" , (pref_cd, pref_name))
pref_cd = 101
pref_name = "Country of ambition"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (%s, %s)""" , (pref_cd, pref_name,))
cnn.commit()
print("Rollback test==========================")
pref_cd = 102
pref_name = "roll back"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (%s, %s)""" , (pref_cd, pref_name,))
cnn.rollback()
print("Examination of stored procedures==========================")
cur.callproc("test_sp", (from_id, to_id))
for rs in cur.stored_results():
print("Recordset...")
rows = rs.fetchall()
for row in rows:
print ("%d %s" % (row[0], row[1]))
print("Examination of stored procedures(Multiple)==================")
cur.callproc("test_sp2", (1, 100))
for rs in cur.stored_results():
print("Recordset...")
rows = rs.fetchall()
for row in rows:
print ("%d %s" % (row[0], row[1]))
print("Function test==========================")
pref_cd = 100
cur.execute("""SELECT test_fn(%s)""" , (pref_cd,))
rows = cur.fetchall()
for row in rows:
print("code:%d name:%s" % (pref_cd, row[0]))
cur.close()
cnn.close()
except (mysql.connector.errors.ProgrammingError) as e:
print (e)
Usability
-MySQL stored is similar to SQL SERVER.
-After executing the stored, the recordset is in cur.stored_results. ..
-MySQL-python has a library, but it doesn't work on 3.x series. The usage is almost the same. http://sourceforge.net/projects/mysql-python/
SQLSERVER
environment
Python2.7 or 3.3 SQLSERVER EXPRESS 2012
Database settings
Allow SQL SERVER connections.
table
CREATE TABLE [dbo].[T01Prefecture](
[PREF_CD] [int] NOT NULL,
[PREF_NAME] [varchar](10) NULL,
PRIMARY KEY CLUSTERED
(
[PREF_CD] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]
__ Stored to return a single recordset __
CREATE PROCEDURE [dbo].[test_sp]
@from INT,
@to INT
AS
BEGIN
-- SET NOCOUNT ON added to prevent extra result sets from
-- interfering with SELECT statements.
SET NOCOUNT ON;
-- Insert statements for procedure here
SELECT * FROM T01Prefecture WHERE PREF_CD BETWEEN @from AND @to
END
__ Stored that returns multiple recordsets __
CREATE PROCEDURE [dbo].[test_sp2]
@cd1 INT,
@cd2 INT
AS
BEGIN
-- SET NOCOUNT ON added to prevent extra result sets from
-- interfering with SELECT statements.
SET NOCOUNT ON;
-- Insert statements for procedure here
SELECT * FROM T01Prefecture WHERE PREF_CD = @cd1
SELECT * FROM T01Prefecture WHERE PREF_CD = @cd2
END
Python code
Operate SQLSERVER directly
pymssql http://pymssql.org/en/latest/
Unstable operation It works for Python2.7 (32bit). You can't even install it with Python3.3 (64bit).
# -*- coding: cp932 -*-
#mssql operation sample
#Stop pymssql because it doesn't encode VARCHAR well.
#easy_install pymssql
import pymssql
cnn = pymssql.connect(host="127.0.0.1\SQLEXPRESS", user="sa", password="xxxx", database="Sample001")
cur = cnn.cursor()
#Organize test data
pref_cd = 100
cur.execute("""DELETE FROM t01prefecture WHERE PREF_CD >= %d"""
% (pref_cd,))
cnn.commit()
print("Simple SELECT statement==========================")
print("Make sure the VARCHAR encoding is wrong")
from_id = 45
to_id = 999
cur.execute("""SELECT PREF_CD, PREF_NAME FROM T01Prefecture
WHERE PREF_CD BETWEEN %d AND %d""" , (from_id, to_id))
rows = cur.fetchall()
for row in rows:
print(row)
# print("%d %s" % (row[0], row[1])) #Get Error
#Character code analysis when handling VARCHAR is invalid
#I am trying to handle the character code of CP932 as UNICODE.
print("Convert to NVARCHAR as it cannot handle VARCHAR fields")
#Since VARCHAR fields cannot be handled, convert to NVARCHAR and return
cur.execute("""SELECT PREF_CD,CAST(PREF_NAME AS NVARCHAR) FROM T01Prefecture
WHERE PREF_CD BETWEEN %d AND %d""" , (from_id, to_id))
rows = cur.fetchall()
for row in rows:
print(row)
print("%d %s" % (row[0], row[1]))
print("Commit test==========================")
pref_cd = 100
pref_name = "Mote Mote country"
cur.execute("""INSERT INTO t01prefecture(PREF_CD, PREF_NAME)
VALUES (%d, %s)""" , (pref_cd, pref_name))
pref_cd = 101
pref_name = "Country of ambition"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (%d, %s)""" , (pref_cd, pref_name,))
cnn.commit()
print("Rollback test==========================")
pref_cd = 102
pref_name = "roll"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (%d, %s)""" , (pref_cd, pref_name,))
cnn.rollback()
cur.execute("""SELECT PREF_CD, CAST(PREF_NAME AS NVARCHAR) FROM t01prefecture
WHERE PREF_CD BETWEEN %d AND %d""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Examination of stored procedures==========================")
cur.execute("""exec test_sp %d, %d """ , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
#Illegal encoding
#print("%d %s" % (row[0], row[1]))
print(row)
print("Stored procedure exams Multiple==========================")
cur.execute("""exec test_sp2 %d, %d """ , (1, 10, ))
for row in rows:
while True:
print ("Recordset...")
rows = cur.fetchall()
for row in rows:
#print("%d %s" % (row[0], row[1]))
print(row)
if not cur.nextset():
break
cur.close()
cnn.close()
Via ODBC
Using __pyodbc: __ https://code.google.com/p/pyodbc/
# -*- coding: cp932 -*-
#mssql operation sample
#easy_install pyodbc
import pyodbc
try:
cnn = pyodbc.connect("DRIVER={SQL Server};SERVER=127.0.0.1\SQLEXPRESS;" +
"UID=sa;PWD=XXXX;DATABASE=Sample001")
cur = cnn.cursor()
#Organize test data
pref_cd = 100
cur.execute("""DELETE FROM t01prefecture WHERE PREF_CD >= ?""", pref_cd)
cnn.commit()
print("Simple SELECT statement==========================")
from_id = 45
to_id = 999
#The following should match the character code of the environment!
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN ? AND ?""" , from_id, to_id)
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Commit test==========================")
pref_cd = 100
pref_name = "Mote Mote country"
cur.execute("""INSERT INTO t01prefecture(PREF_CD, PREF_NAME)
VALUES (?, ?)""" , pref_cd, pref_name)
pref_cd = 101
pref_name = "Country of ambition"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (?, ?)""" , pref_cd, pref_name)
cnn.commit()
print("Rollback test==========================")
pref_cd = 102
pref_name = "roll"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (?, ?)""" , pref_cd, pref_name)
cnn.rollback()
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN ? AND ?""" , from_id, to_id)
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Examination of stored procedures==========================")
cur.execute("""exec test_sp ?, ? """ , from_id, to_id)
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Stored procedure exams Multiple====================")
cur.execute("""exec test_sp2 ?, ? """ , 1, 10)
while True:
print ("Recordset...")
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
if not cur.nextset():
break
cur.close()
cnn.close()
except (pyodbc.Error) as e:
print (e)
print (e.args[1])
Example of using __pypyodbc: __ https://code.google.com/p/pypyodbc/
# -*- coding: cp932 -*-
#mssql operation sample
#easy_install pypyodbc
import pypyodbc
try:
cnn = pypyodbc.connect("DRIVER={SQL Server};SERVER=127.0.0.1\SQLEXPRESS;UID=sa;PWD=xxxxx;DATABASE=Sample001")
cur = cnn.cursor()
#Organize test data
pref_cd = 100
cur.execute("""DELETE FROM t01prefecture WHERE PREF_CD >= ?"""
, (pref_cd,))
cnn.commit()
print("Simple SELECT statement==========================")
from_id = 45
to_id = 999
#The following should match the character code of the environment!
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN ? AND ?""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Commit test==========================")
pref_cd = 100
pref_name = "Mote Mote country"
cur.execute("""INSERT INTO t01prefecture(PREF_CD, PREF_NAME)
VALUES (?, ?)""" , (pref_cd, pref_name))
pref_cd = 101
pref_name = "Country of ambition"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (?, ?)""" , (pref_cd, pref_name,))
cnn.commit()
print("Rollback test==========================")
pref_cd = 102
pref_name = "roll"
cur.execute("""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (?, ?)""" , (pref_cd, pref_name,))
cnn.rollback()
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN ? AND ?""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Examination of stored procedures==========================")
cur.execute("""exec test_sp ?, ? """ , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
print("Stored procedure exams Multiple==========================")
cur.execute("""exec test_sp2 ?, ? """ , (1, 10, ))
while True:
print ("Recordset...")
rows = cur.fetchall()
for row in rows:
print("%d %s" % (row[0], row[1]))
if not cur.nextset():
break
cur.close()
cnn.close()
except (pypyodbc.DatabaseError) as e:
print (e.args[1])
Usability
pymssql can be used without going through ODBC, but the operation is extremely unstable. The behavior of VARCHAR is suspicious and it cannot be used unless it is cast to NVARCHAR. Also, you can't even install it on 64-bit Python 3.3.
All libraries via ODBC worked. I have not confirmed it except for Windows, but it is said that it supports it.
ORACLE
environment
Python 2.7 or 3.3 Oracle11
- Install the Oracle client. At this time, create an environment for developers.
- Download and install from the following, or easy_install. cx_Oracle http://cx-oracle.sourceforge.net/
Python code
# -*- coding: cp932 -*-
# cx_ORACLE operation in Python using Oracle
#1.Install the Oracle client.
#At this time, create an environment for developers.
#(Maybe you need OCI?)
#
#2.Download and install from below or easy_install.
# cx_Oracle
# http://cx-oracle.sourceforge.net/
import cx_Oracle
import os
os.environ["NLS_LANG"] = "JAPANESE_JAPAN.JA16SJISTILDE"
try:
tns = cx_Oracle.makedsn("localhost", 1521, "Sample")
conn = cx_Oracle.connect("user", "pass", tns)
cur = conn.cursor()
from_cd = 45
to_cd = 200
cur.execute("""SELECT * FROM T01PREFECTURE
WHERE PREF_CD BETWEEN :arg1 AND :arg2""",
arg1=from_cd,
arg2=to_cd)
rows = cur.fetchall()
for r in rows:
print("%d : %s" % (r[0], r[1]))
print ("===================")
print ("Erase 100")
cur.execute("""DELETE FROM T01PREFECTURE WHERE PREF_CD=:arg1""",
arg1=100)
cur.execute("""SELECT * FROM T01PREFECTURE
WHERE PREF_CD BETWEEN :arg1 AND :arg2""",
arg1=from_cd,
arg2=to_cd)
rows = cur.fetchall()
for r in rows:
print("%d : %s" % (r[0], r[1]))
print ("------------------")
print ("Add 100")
cur.execute("""INSERT INTO T01PREFECTURE
VALUES (:arg1, :arg2)""",
arg1=100,
arg2="I")
conn.commit()
cur.execute("""SELECT * FROM T01PREFECTURE
WHERE PREF_CD BETWEEN :arg1 AND :arg2""",
arg1=from_cd,
arg2=to_cd)
rows = cur.fetchall()
for r in rows:
print("%d : %s" % (r[0], r[1]))
print ("===================")
print ("101 added")
cur.execute("""INSERT INTO T01PREFECTURE
VALUES (:arg1, :arg2)""",
arg1=101,
arg2="Hokuto")
cur.execute("""SELECT * FROM T01PREFECTURE
WHERE PREF_CD BETWEEN :arg1 AND :arg2""",
arg1=from_cd,
arg2=to_cd)
rows = cur.fetchall()
for r in rows:
print("%d : %s" % (r[0], r[1]))
print ("------------------")
print ("roll back")
conn.rollback()
cur.execute("""SELECT * FROM T01PREFECTURE
WHERE PREF_CD BETWEEN :arg1 AND :arg2""",
arg1=from_cd,
arg2=to_cd)
rows = cur.fetchall()
for r in rows:
print("%d : %s" % (r[0], r[1]))
except (cx_Oracle.DatabaseError) as ex:
error, = ex.args
print (error.message)
Usability
It seems that you are operating via OCI, so you have to install the ORACLE client. I have a habit of connecting. Also, specify the character code with environment variables.
PL / SQL in ORACLE, unlike SQL SERVER, does not return a result set. (I can return an array, but I haven't experimented here because it's a hassle)
If you don't explicitly COMMIT, your changes will be discarded.
Postgresql
environment
Python 2.x / 3.x PostgresSQL 9.3
Use psycopg2 as a library For Windows, download from the following page and execute Exe http://www.stickpeople.com/projects/python/win-psycopg/
Database settings
table definition
CREATE TABLE t01prefecture
(
pref_cd integer NOT NULL,
pref_name character varying(10),
CONSTRAINT t01prefecture_pkey PRIMARY KEY (pref_cd)
)
WITH (
OIDS=FALSE
);
ALTER TABLE t01prefecture
OWNER TO postgres;
__Postgres SQL user-defined function __
CREATE OR REPLACE FUNCTION test_sp(IN from_cd integer, IN to_cd integer)
RETURNS TABLE(code integer, name varchar) AS
$$
DECLARE
BEGIN
RETURN QUERY SELECT PREF_CD,PREF_NAME FROM t01Prefecture
WHERE PREF_CD BETWEEN from_cd AND to_cd;
END;
$$ LANGUAGE plpgsql;
Python code
# -*- coding: cp932 -*-
#For Windows, download from the following
# http://www.stickpeople.com/projects/python/win-psycopg/
# Python3.For x, unicode(row[1],'utf-8')Is unnecessary.
#
import psycopg2
try:
cnn = psycopg2.connect("dbname=Sample001 host=localhost user=postgres password=xxxxx")
cur = cnn.cursor()
#Organize test data
pref_cd = 100
cur.execute("""DELETE FROM t01prefecture WHERE PREF_CD >= %s"""
, (pref_cd,))
cnn.commit()
print("Simple SELECT statement==========================")
from_id = 45
to_id = 999
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN %s AND %s""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
#print("%d %s" % (row[0], unicode(row[1],'utf-8')))
print("%d %s" % (row[0], row[1]))
print("Commit test==========================")
pref_cd = 100
pref_name = u"Mote Mote country"
cur.execute(u"""INSERT INTO t01prefecture(PREF_CD, PREF_NAME)
VALUES (%s, %s)""" , (pref_cd, pref_name))
pref_cd = 101
pref_name = u"Country of ambition"
cur.execute(u"""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (%s, %s)""" , (pref_cd, pref_name,))
cnn.commit()
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN %s AND %s""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
#print("%d %s" % (row[0],unicode(row[1],'utf-8')))
print("%d %s" % (row[0],row[1]))
print("Rollback test==========================")
pref_cd = 102
pref_name = u"roll"
cur.execute(u"""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (%s, %s)""" , (pref_cd, pref_name,))
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN %s AND %s""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
#print("%d %s" % (row[0], unicode(row[1],'utf-8')))
print("%d %s" % (row[0], row[1]))
print("-------------------------")
cnn.rollback()
cur.execute("""SELECT PREF_CD,PREF_NAME FROM t01prefecture
WHERE PREF_CD BETWEEN %s AND %s""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
#print("%d %s" % (row[0], unicode(row[1],'utf-8')))
print("%d %s" % (row[0], row[1]))
print("User-defined==========================")
cur.execute("""SELECT * FROM test_sp(%s,%s)""" , (from_id, to_id, ))
rows = cur.fetchall()
for row in rows:
#print("%d %s" % (row[0], unicode(row[1],'utf-8')))
print("%d %s" % (row[0], row[1]))
cur.close()
cnn.close()
except (psycopg2.OperationalError) as e:
print (e)
Usability
Python2.X returns the recordset in UTF8, so you have to convert it to UNICODE once. No need for Python3 system.
Postgres stored procedures are similar to ORACLE's PL / SQL. However, you can specify a function that returns a table type.
SQLite
environment
no special instruction. If you have Python 2.5 or later, you should be able to operate SQLITE3 as standard.
Python code
# -*- coding: cp932 -*-
#sqlite3 is Python2.Should be standard from 5 onwards.
import sqlite3
conn = sqlite3.connect('test.sqlite3')
sql = '''CREATE TABLE IF NOT EXISTS t01prefecture(
pref_cd INTEGER,
pref_name TEXT);'''
conn.execute(sql)
conn.execute(u"DELETE FROM t01prefecture")
#Commit test
pref_cd = 100
pref_name = u"Mote Mote country"
conn.execute(u"""INSERT INTO t01prefecture(PREF_CD, PREF_NAME)
VALUES (?, ?)""" , (pref_cd, pref_name))
pref_cd = 101
pref_name = u"Country of ambition"
conn.execute(u"""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (?, ?)""" , (pref_cd, pref_name,))
conn.commit()
#Rollback test
pref_cd = 102
pref_name = u"back"
conn.execute(u"""INSERT INTO t01prefecture(PREF_CD,PREF_NAME)
VALUES (?, ?)""" , (pref_cd, pref_name,))
conn.rollback()
rows = conn.execute(u'SELECT * FROM t01prefecture WHERE pref_cd > ?', (0,))
for row in rows:
print(u"%d %s" % (row[0], row[1]))
#User-defined
#Only concatenate characters
class UserDef:
def __init__(self):
self.values = []
def step(self, value):
self.values.append(value)
def finalize(self):
return "/".join(map(str, self.values))
conn.create_aggregate("userdef", 1, UserDef)
rows = conn.execute(u'SELECT userdef(PREF_NAME) FROM t01prefecture')
for row in rows:
print(u"%s" % (row[0]))
conn.close()
Usability
Unlike conventional DBs, there is no need to build a server. User-defined functions can be set on the client side.
Recommended Posts