本文转至 余子越的博客 ,文章 Python 连接 MySQL,欢迎访问yuchaoshui.com 了解更多信息!
一、普通 MySQL 连接方法
??使用模块 MySQLdb 普通方式连接。
#!/usr/bin/env python
# _*_ coding:utf-8 _*_
import MySQLdb
conn = MySQLdb.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘test‘)
cursor = conn.cursor()
sql_1 = "select * from user where id = %s;" % (5,)
sql_2 = "select * from user where id = %s;" % (5,)
sql_3 = """
insert into user(username, password)
values("yuchaoshui", "123");
"""
try:
print cursor.execute(sql_1)
print cursor.fetchall()
print cursor.execute(sql_2)
print cursor.fetchall()
print cursor.execute(sql_3)
conn.commit()
except Exception as e:
print(e)
conn.rollback()
cursor.close()
conn.close()
?? execute()
返回结果表示影响的行数。cursor.fetchone()
取回一条结果。sql_1 直接一行写完,sql_2 换行写完, sql_3 多行写。 查询时不需要 commit()
操作,插入、更新、删除时需要 commit()
提交。
二、使用连接池连接MySQL
#!/usr/bin/env python
# _*_ coding:utf-8 _*_
import MySQLdb
from DBUtils.PooledDB import PooledDB
pool = PooledDB(MySQLdb, 5, host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘test‘)
conn = pool.connection()
cursor = conn.cursor()
sql_1 = "select * from user where id = %s;" % (5,)
sql_2 = "select * from user where id = %s;" % (5,)
sql_3 = """
insert into user(username, password)
values("yuchaoshui", "123");
"""
try:
print cursor.execute(sql_1)
print cursor.fetchall()
print cursor.execute(sql_2)
print cursor.fetchall()
print cursor.execute(sql_3)
conn.commit()
except Exception as e:
print(e)
conn.rollback()
cursor.close()
conn.close()
?? 5 为连接池里的最少连接数, 以后每次需要数据库连接就是用connection()
函数获取连接就好了
PooledDB 的默认值
PooledDB(self, creator, mincached=0, maxcached=0, maxshared=0, maxconnections=0, blocking=False, maxusage=None, setsession=None, reset=True, failures=None, ping=1, *args, **kwargs)
- PooledDB的参数:
- mincached,最少的空闲连接数,如果空闲连接数小于这个数,pool会创建一个新的连接
- maxcached,最大的空闲连接数,如果空闲连接数大于这个数,pool会关闭空闲连接
- maxconnections,最大的连接数,
- blocking,当连接数达到最大的连接数时,在请求连接的时候,如果这个值是True,请求连接的程序会一直等待,直到当前连接数小于最大连接数,如果这个值是False,会报错,
- maxshared , 当连接数达到这个数,新请求的连接会分享已经分配出去的连接
三、模块导入连接 MySQL
??以连接池的方式,编写模块 mysqlhelper.py,可以在项目的其他地方导入MySQL连接实例即可使用。 模块点此下载 mysqlhelper.py
#!/usr/bin/env python
# _*_ coding:utf-8 _*_
from __future__ import print_function
from DBUtils.PooledDB import PooledDB
import MySQLdb
import sys
__all__ = [‘m‘] + ["m"+str(i) for i in range(2, 11)]
class MH(object):
def __init__(self):
try:
print("Connecting MySQL Server {0}@{1}:{2} ..".format(
self.__class__.db, self.__class__.host, self.__class__.port), end=‘.‘)
self.conn = self.__class__.pool.connection()
self.cursor = self.conn.cursor()
print(‘ ok!‘)
except Exception, e:
print("pool.connection error: {0}".format(e))
def select(self, query=‘‘):
try:
self.effect = self.cursor.execute(query)
return self.cursor
except Exception as e:
print("select error: {0}".format(e))
def update(self, query=‘‘):
try:
self.effect = self.cursor.execute(query)
self.conn.commit()
except Exception as e:
print("update error: {0}".format(e))
self.conn.rollback()
self.effect = 0
# M2 类继承自 M1,表示一个新的 MySQL 连接池。
# 如果需要新的连接池 ,按照如下格式新增即可。
class MH2(MH):
pass
def init_pool(M,
host=‘127.0.0.1‘,
port=3306,
user=‘root‘,
password=‘‘,
database=‘test‘,
pool_size=5):
M.host = host
M.port = int(port)
M.user = user
M.password = password
M.db = database
M.pool_size = pool_size
try:
M.pool = PooledDB(MySQLdb,
M.pool_size,
host=M.host,
port=M.port,
user=M.user,
passwd=M.password,
db=M.db)
except Exception, e:
print("PooledDB init error: {0}".format(e))
exit(1)
# 初始化连接池,可以有多个。第一个参数是前面手动定义的连接池类。
init_pool(MH, ‘127.0.0.1‘, 3306, ‘root‘, ‘123‘, ‘test‘)
init_pool(MH2, ‘12.55.5.61‘, 3306, ‘root‘, ‘123‘, ‘test‘)
# 定义将要被导出的MySQL实例。 一个连接池可同时提供多个实例对象。
m = MH()
m2 = MH2()
if __name__ == "__main__":
pass
#print "\nm info:"
#print m.select("select * from user;").fetchone()
#print m.effect
#print m.select("select * from user;").fetchall()
#print m.effect
#m.update("insert into user(username,password) values(‘haha‘, ‘heihei‘);")
#print m.effect
##################################################
#print "\nm2 info:"
#print m2.select("select * from user;").fetchone()
#print m2.effect
#print m2.select("select * from user;").fetchall()
#print m2.effect
#m2.update("insert into user(username,password) values(‘haha‘, ‘heihei‘);")
#print m2.effect
- 使用方法
#!/usr/bin/env python
# _*_ coding:utf-8 _*_
from mysqlhelper import m, m2
import time
def test():
print "\nm info:"
print m.select("select * from user;").fetchone()
print m.effect
print m.select("select * from user;").fetchall()
print m.effect
m.update("insert into user(username,password) values(‘haha‘, ‘heihei‘);")
print m.effect
#################################################
print "\nm2 info:"
print m2.select("select * from user;").fetchone()
print m2.effect
print m2.select("select * from user;").fetchall()
print m2.effect
m2.update("insert into user(username,password) values(‘haha‘, ‘heihei‘);")
print m2.effect
if __name__ == ‘__main__‘:
test()
本文转至 余子越的博客 ,文章 Python 连接 MySQL,欢迎访问yuchaoshui.com 了解更多信息!