python 连接各类主流数据库简单示例【转载】

本篇博文主要介绍Python连接各种数据库的方法及简单使用
包括关系数据库：sqlite,mysql,mssql
非关系数据库：MongoDB,Redis

代码写的比较清楚，直接上代码

1.连接sqlite

# coding=utf-8
# http://www.runoob.com/sqlite/sqlite-python.html import sqlite3
import traceback

try:
# 如果表不存在,就创建
with sqlite3.connect('test.db') as conn:

print("Opened database successfully")

# 删除表
conn.execute("DROP TABLE IF EXISTS  COMPANY")

# 创建表
sql = """
CREATE TABLE IF NOT EXISTS COMPANY
(ID INTEGER  PRIMARY KEY       AUTOINCREMENT,
NAME           TEXT    NOT NULL,
AGE            INT     NOT NULL,
ADDRESS        CHAR(50),
SALARY         REAL);
"""
conn.execute(sql)

print("create table successfully")

# 添加数据
conn.executemany("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY) VALUES (?, ?, ?, ? )",
[('Paul', 32, 'California', 20000.00),
('Allen', 25, 'Texas', 15000.00),
('Teddy', 23, 'Norway', 20000.00),
('Mark', 25, 'Rich-Mond ', 65000.00),
('David', 27, 'Texas', 85000.00),
('Kim', 22, 'South-Hall', 45000.00),
('James', 24, 'Houston', 10000.00)])
# conn.execute("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY)\
# VALUES ( 'Paul', 32, 'California', 20000.00 )")
#
# conn.execute("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY)\
# VALUES ('Allen', 25, 'Texas', 15000.00 )")
#
# conn.execute("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY)\
# VALUES ('Teddy', 23, 'Norway', 20000.00 )")
#
# conn.execute("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY)\
# VALUES ( 'Mark', 25, 'Rich-Mond ', 65000.00 )")
#
# conn.execute("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY)\
# VALUES ( 'David', 27, 'Texas', 85000.00 )");
#
# conn.execute("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY)\
# VALUES ( 'Kim', 22, 'South-Hall', 45000.00 )")
#
# conn.execute("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY)\
# VALUES ( 'James', 24, 'Houston', 10000.00 )")

# 提交,否则重新运行程序时,表中无数据
conn.commit()
print("insert successfully")

# 查询表
sql = """
select id,NAME,AGE,ADDRESS,SALARY FROM COMPANY
"""

result = conn.execute(sql)

for row in result:
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("id", row[0]))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row[1]))
print("%-10s %s" % ("age", row[2]))
print("%-10s %s" % ("address", row[3]))
print("%-10s %.2f" % ("salary", row[4]))
# or
# print('{:10s} {:.2f}'.format("salary", row[4]))

except sqlite3.Error as e:
print("sqlite3 Error:", e)
traceback.print_exc()

2.连接mysql

2.1使用mysqldb库中的_mysql

#! /usr/bin/env python2.7
# coding=utf-8
# Created by xiaosanyu at 16/5/30

# mysqldb 只支持python2.7
# http://mysql-python.sourceforge.net/ 
import MySQLdb
from contextlib import closing
import traceback

try:
# 获取一个数据库连接
with closing(MySQLdb.connect(host='localhost', user='root', passwd='root', db='test', port=3306,charset='utf8')) as conn:
print("connect database successfully")
with closing(conn.cursor()) as cur:
# 删除表
cur.execute("DROP TABLE IF EXISTS  COMPANY")
# 创建表
sql = """
CREATE TABLE IF NOT EXISTS COMPANY
(ID INTEGER  PRIMARY KEY NOT NULL  auto_increment,
NAME           TEXT    NOT NULL,
AGE            INT     NOT NULL,
ADDRESS        CHAR(50),
SALARY         REAL);
"""
cur.execute(sql)

print("create table successfully")

# 添加数据
# 在一个conn.execute里面里面执行多个sql语句是非法的
cur.executemany("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY) VALUES ( %s, %s, %s, %s )",
[('Paul', 32, 'California', 20000.00),
('Allen', 25, 'Texas', 15000.00),
('Teddy', 23, 'Norway', 20000.00),
('Mark', 25, 'Rich-Mond ', 65000.00),
('David', 27, 'Texas', 85000.00),
('Kim', 22, 'South-Hall', 45000.00),
('James', 24, 'Houston', 10000.00)])

# 提交,否则重新运行程序时,表中无数据
conn.commit()
print("insert successfully")

# 查询表
sql = """
select id,NAME,AGE,ADDRESS,SALARY FROM COMPANY
"""

cur.execute(sql)

for row in cur.fetchall():
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("id", row[0]))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row[1]))
print("%-10s %s" % ("age", row[2]))
print("%-10s %s" % ("address", row[3]))
print("%-10s %s" % ("salary", row[4]))

except MySQLdb.Error as e:
print("Mysql Error:", e)
traceback.print_exc()  # 打印错误栈信息

2.2 使用MySQLdb

#! /usr/bin/env python2.7
# coding=utf-8
# Created by xiaosanyu at 16/5/30

# mysqldb 只支持python2.7
# http://mysql-python.sourceforge.net/ 
import MySQLdb
from contextlib import closing
import traceback

try:
# 获取一个数据库连接
with closing(MySQLdb.connect(host='localhost', user='root', passwd='root', db='test', port=3306,charset='utf8')) as conn:
print("connect database successfully")
with closing(conn.cursor()) as cur:
# 删除表
cur.execute("DROP TABLE IF EXISTS  COMPANY")
# 创建表
sql = """
CREATE TABLE IF NOT EXISTS COMPANY
(ID INTEGER  PRIMARY KEY NOT NULL  auto_increment,
NAME           TEXT    NOT NULL,
AGE            INT     NOT NULL,
ADDRESS        CHAR(50),
SALARY         REAL);
"""
cur.execute(sql)

print("create table successfully")

# 添加数据
# 在一个conn.execute里面里面执行多个sql语句是非法的
cur.executemany("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY) VALUES ( %s, %s, %s, %s )",
[('Paul', 32, 'California', 20000.00),
('Allen', 25, 'Texas', 15000.00),
('Teddy', 23, 'Norway', 20000.00),
('Mark', 25, 'Rich-Mond ', 65000.00),
('David', 27, 'Texas', 85000.00),
('Kim', 22, 'South-Hall', 45000.00),
('James', 24, 'Houston', 10000.00)])

# 提交,否则重新运行程序时,表中无数据
conn.commit()
print("insert successfully")

# 查询表
sql = """
select id,NAME,AGE,ADDRESS,SALARY FROM COMPANY
"""

cur.execute(sql)

for row in cur.fetchall():
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("id", row[0]))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row[1]))
print("%-10s %s" % ("age", row[2]))
print("%-10s %s" % ("address", row[3]))
print("%-10s %s" % ("salary", row[4]))

except MySQLdb.Error as e:
print("Mysql Error:", e)
traceback.print_exc()  # 打印错误栈信息

2.3使用pymysql

2.1和2.2节使用MySQLdb，不支持Python3.x
pymysql对Python2.x和Python3.x的支持都比较好

# Created by xiaosanyu at 16/5/30
# coding=utf-8

# https://github.com/PyMySQL/PyMySQL/ import pymysql
from contextlib import closing
import traceback

try:
# 获取一个数据库连接,with关键字 表示退出时,conn自动关闭
# with 嵌套上一层的with 要使用closing()
with closing(pymysql.connect(host='localhost', user='root', passwd='root', db='test', port=3306,
charset='utf8')) as conn:

print("connect database successfully")

# 获取游标,with关键字 表示退出时,cur自动关闭
with conn.cursor() as cur:
# 删除表
cur.execute("DROP TABLE IF EXISTS  COMPANY")
# 创建表
sql = """
CREATE TABLE IF NOT EXISTS COMPANY
(ID INTEGER  PRIMARY KEY NOT NULL  auto_increment,
NAME           TEXT    NOT NULL,
AGE            INT     NOT NULL,
ADDRESS        CHAR(50),
SALARY         REAL);
"""
cur.execute(sql)

print("create table successfully")

# 添加数据
# 在一个conn.execute里面里面执行多个sql语句是非法的
cur.executemany("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY) VALUES ( %s, %s, %s, %s )",
[('Paul', 32, 'California', 20000.00),
('Allen', 25, 'Texas', 15000.00),
('Teddy', 23, 'Norway', 20000.00),
('Mark', 25, 'Rich-Mond ', 65000.00),
('David', 27, 'Texas', 85000.00),
('Kim', 22, 'South-Hall', 45000.00),
('James', 24, 'Houston', 10000.00)])

# 提交,否则重新运行程序时,表中无数据
conn.commit()
print("insert successfully")

# 查询表
sql = """
select id,NAME,AGE,ADDRESS,SALARY FROM COMPANY
"""

cur.execute(sql)

for row in cur.fetchall():
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("id", row[0]))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row[1]))
print("%-10s %s" % ("age", row[2]))
print("%-10s %s" % ("address", row[3]))
print("%-10s %s" % ("salary", row[4]))
except pymysql.Error as e:
print("Mysql Error:", e)
traceback.print_exc()

3.连接mssql

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import pymssql

class MSSQL(object):
'''
对pymssql的简单封装
pymssql库，该库到这里下载：http://www.lfd.uci.edu/~gohlke/pythonlibs/#pymssql
使用该库时，需要在Sql Server Configuration Manager里面将TCP/IP协议开启

用法：
'''

def __init__(self,host,user,pwd,db):
self.host=host
self.user=user
self.pwd=pwd
self.db=db

def GetConnect(self):
'''
得到链接信息
:return:
'''
if not self.db:
raise (NameError,"没有设置数据库信息")
self.connect=pymssql.connect(host=self.host,user=self.user,password=self.pwd,database=self.db,charset="utf8")
cur=self.connect.cursor()
if not cur:
raise (NameError,"链接数据库失败")
else:
return cur

def ExecQuery(self,sql):
'''
执行查询语句
返回的是一个包含tuple的list，list的元素是记录行，tuple的元素是每行记录的字段

调用示例：
ms = MSSQL(host="localhost",user="sa",pwd="123456",db="PythonWeiboStatistics")
resList = ms.ExecQuery("SELECT id,NickName FROM WeiBoUser")
for (id,NickName) in resList:
print str(id),NickName
:param sql: sql语句
:return:
'''
cur=self.GetConnect()
cur.execute(sql)
resList = cur.fetchall()

#查询完毕后必须关闭连接
self.connect.close()
return resList

def ExecNonQuery(self, sql):
"""
执行非查询语句

调用示例：
cur = self.__GetConnect()
cur.execute(sql)
self.conn.commit()
self.conn.close()
"""
cur = self.__GetConnect()
cur.execute(sql)
self.connect.commit()
self.connect.close()

def test():
## ms = MSSQL(host="localhost",user="sa",pwd="123456",db="PythonWeiboStatistics")
## #返回的是一个包含tuple的list，list的元素是记录行，tuple的元素是每行记录的字段
## ms.ExecNonQuery("insert into WeiBoUser values('2','3')")

ms = MSSQL(host="112.74.160.198", user="zl", pwd="zl@123456", db="zhongla20170901")
resList = ms.ExecQuery("SELECT * FROM Pub_Law")
print(resList)

if __name__ == '__main__':
test()

# Created by xiaosanyu at 16/5/30

# http://www.pymssql.org/en/latest/ import pymssql
from contextlib import closing

try:
# 先要保证数据库中有test数据库
# 获取一个数据库连接,with关键字 表示退出时,conn自动关闭
# with 嵌套上一层的with 要使用closing()
with closing(pymssql.connect(host='192.168.100.114', user='sa', password='sa12345', database='test', port=1433,
charset='utf8')) as conn:

print("connect database successfully")

# 获取游标,with关键字 表示退出时,cur自动关闭
with conn.cursor() as cur:
# 删除表
cur.execute(
'''if exists (select 1 from  sys.objects where name='COMPANY' and  type='U')  drop table COMPANY''')
# 创建表
sql = """
CREATE TABLE  COMPANY
(ID INT  IDENTITY(1,1) PRIMARY KEY NOT NULL ,
NAME           TEXT    NOT NULL,
AGE            INT     NOT NULL,
ADDRESS        CHAR(50),
SALARY         REAL);
"""
cur.execute(sql)

print("create table successfully")

# 添加数据
# 在一个conn.execute里面里面执行多个sql语句是非法的
cur.executemany("INSERT INTO COMPANY (NAME,AGE,ADDRESS,SALARY) VALUES ( %s, %s, %s, %s )",
[('Paul', 32, 'California', 20000.00),
('Allen', 25, 'Texas', 15000.00),
('Teddy', 23, 'Norway', 20000.00),
('Mark', 25, 'Rich-Mond', 65000.00),
('David', 27, 'Texas', 85000.00),
('Kim', 22, 'South-Hall', 45000.00),
('James', 24, 'Houston', 10000.00)])

# 提交,否则重新运行程序时,表中无数据
conn.commit()
print("insert successfully")

# 查询表
sql = """
select id,NAME,AGE,ADDRESS,SALARY FROM COMPANY
"""

cur.execute(sql)

for row in cur.fetchall():
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("id", row[0]))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row[1]))
print("%-10s %s" % ("age", row[2]))
print("%-10s %s" % ("address", row[3]))
print("%-10s %s" % ("salary", row[4]))
except pymssql.Error as e:
print("mssql Error:", e)
# traceback.print_exc()

4.连接MongoDB

# Created by xiaosanyu at 16/5/30

# https://docs.mongodb.com/ecosystem/drivers/python/ # https://pypi.python.org/pypi/pymongo/ 
import pymongo
from pymongo.mongo_client import MongoClient
import pymongo.errors
import traceback

try:
# 连接到 mongodb 服务
mongoClient = MongoClient('localhost', 27017)
# 连接到数据库
mongoDatabase = mongoClient.test
print("connect database successfully")

# 获取集合
mongoCollection = mongoDatabase.COMPANY

# 移除所有数据
mongoCollection.remove()

# 添加数据
mongoCollection.insert_many([{"Name": "Paul", "Age": "32", "Address": "California", "Salary": "20000.00"},
{"Name": "Allen", "Age": "25", "Address": "Texas", "Salary": "15000.00"},
{"Name": "Teddy", "Age": "23", "Address": "Norway", "Salary": "20000.00"},
{"Name": "Mark", "Age": "25", "Address": "Rich-Mond", "Salary": "65000.00"},
{"Name": "David", "Age": "27", "Address": "Texas", "Salary": "85000.00"},
{"Name": "Kim", "Age": "22", "Address": "South-Hall", "Salary": "45000.00"},
{"Name": "James", "Age": "24", "Address": "Houston", "Salary": "10000.00"}, ])

#获取集合中的值
for row in mongoCollection.find():
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("_id", row['_id']))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row['Name']))
print("%-10s %s" % ("age", row['Age']))
print("%-10s %s" % ("address", row['Address']))
print("%-10s %s" % ("salary", row['Salary']))

print('\n\n\n')
# 使id自增
mongoCollection.remove()
# 创建计数表
mongoDatabase.counters.save({"_id": "people_id", "sequence_value": 0})
# 创建存储过程
mongoDatabase.system_js.getSequenceValue = '''function getSequenceValue(sequenceName){
var sequenceDocument = db.counters.findAndModify({
query: {_id: sequenceName},
update: {$inc:{sequence_value: 1}},
new:true
});
return sequenceDocument.sequence_value;
}'''
mongoCollection.insert_many(
[{"_id": mongoDatabase.eval("getSequenceValue('people_id')"), "Name": "Paul", "Age": "32",
"Address": "California", "Salary": "20000.00"},
{"_id": mongoDatabase.eval("getSequenceValue('people_id')"), "Name": "Allen", "Age": "25",
"Address": "Texas", "Salary": "15000.00"},
{"_id": mongoDatabase.eval("getSequenceValue('people_id')"), "Name": "Teddy", "Age": "23",
"Address": "Norway", "Salary": "20000.00"},
{"_id": mongoDatabase.eval("getSequenceValue('people_id')"), "Name": "Mark", "Age": "25",
"Address": "Rich-Mond", "Salary": "65000.00"},
{"_id": mongoDatabase.eval("getSequenceValue('people_id')"), "Name": "David", "Age": "27",
"Address": "Texas", "Salary": "85000.00"},
{"_id": mongoDatabase.eval("getSequenceValue('people_id')"), "Name": "Kim", "Age": "22",
"Address": "South-Hall", "Salary": "45000.00"},
{"_id": mongoDatabase.eval("getSequenceValue('people_id')"), "Name": "James", "Age": "24",
"Address": "Houston", "Salary": "10000.00"}, ])

for row in mongoCollection.find():
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("_id", int(row['_id'])))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row['Name']))
print("%-10s %s" % ("age", row['Age']))
print("%-10s %s" % ("address", row['Address']))
print("%-10s %s" % ("salary", row['Salary']))
except pymongo.errors.PyMongoError as e:
print("mongo Error:", e)
traceback.print_exc()

5.连接Redis

5.1使用redis

# coding=utf-8
# Created by xiaosanyu at 16/5/31

# https://pypi.python.org/pypi/redis/2.10.5 # http://redis-py.readthedocs.io/en/latest/# import redis

r = redis.Redis(host='localhost', port=6379, db=0, password="12345")
print("connect", r.ping())

# 看信息
info = r.info()
# or 查看部分信息
# info = r.info("Server")

# 输出信息
items = info.items()
for i, (key, value) in enumerate(items):
print("item %s----%s:%s" % (i, key, value))

# 删除键和对应的值
r.delete("company")

# 可以一次性push一条或多条数据
r.rpush("company", {"id": 1, "Name": "Paul", "Age": "32", "Address": "California", "Salary": "20000.00"},
{"id": 2, "Name": "Allen", "Age": "25", "Address": "Texas", "Salary": "15000.00"},
{"id": 3, "Name": "Teddy", "Age": "23", "Address": "Norway", "Salary": "20000.00"})
r.rpush("company", {"id": 4, "Name": "Mark", "Age": "25", "Address": "Rich-Mond", "Salary": "65000.00"})
r.rpush("company", {"id": 5, "Name": "David", "Age": "27", "Address": "Texas", "Salary": "85000.00"})
r.rpush("company", {"id": 6, "Name": "Kim", "Age": "22", "Address": "South-Hall", "Salary": "45000.00"})
r.rpush("company", {"id": 7, "Name": "James", "Age": "24", "Address": "Houston", "Salary": "10000.00"})

# eval用来将dict格式的字符串转换成dict
for row in map(lambda x: eval(x), r.lrange("company", 0, r.llen("company"))):
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("_id", row['id']))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row['Name']))
print("%-10s %s" % ("age", row['Age']))
print("%-10s %s" % ("address", row['Address']))
print("%-10s %s" % ("salary", row['Salary']))

# 关闭当前连接
# r.shutdown() #这个是关闭redis服务端

5.2使用pyredis

# Created by xiaosanyu at 16/5/30

# http://pyredis.readthedocs.io/en/latest/ import pyredis

r = pyredis.Client(host='localhost', port=6379, database=0, password="12345")
print("connect", r.ping().decode("utf-8"))

# 看信息

# info = r.execute("info").decode()
# or 查看部分信息
info = r.execute("info", "Server").decode()

# 输出信息
print(info)

# 删除键和对应的值
r.delete("company")

# 可以一次性push一条或多条数据
r.rpush("company", '''{"id": 1, "Name": "Paul", "Age": "32", "Address": "California", "Salary": "20000.00"}''',
'''{"id": 2, "Name": "Allen", "Age": "25", "Address": "Texas", "Salary": "15000.00"}''',
'''{"id": 3, "Name": "Teddy", "Age": "23", "Address": "Norway", "Salary": "20000.00"}''')
r.rpush("company", '''{"id": 4, "Name": "Mark", "Age": "25", "Address": "Rich-Mond", "Salary": "65000.00"}''')
r.rpush("company", '''{"id": 5, "Name": "David", "Age": "27", "Address": "Texas", "Salary": "85000.00"}''')
r.rpush("company", '''{"id": 6, "Name": "Kim", "Age": "22", "Address": "South-Hall", "Salary": "45000.00"}''')
r.rpush("company", '''{"id": 7, "Name": "James", "Age": "24", "Address": "Houston", "Salary": "10000.00"}''')

# eval用来将dict格式的字符串转换成dict
for row in map(lambda x: eval(x), r.lrange("company", 0, r.llen("company"))):
print("-" * 50)  # 输出50个-,作为分界线
print("%-10s %s" % ("_id", row['id']))  # 字段名固定10位宽度,并且左对齐
print("%-10s %s" % ("name", row['Name']))
print("%-10s %s" % ("age", row['Age']))
print("%-10s %s" % ("address", row['Address']))
print("%-10s %s" % ("salary", row['Salary']))

# 关闭当前连接
r.close()