脚本宝典收集整理的这篇文章主要介绍了使用python连接mysql数据库,脚本宝典觉得挺不错的,现在分享给大家,也给大家做个参考。
安装数据库驱动:pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple 安装连接池库:pip install dbutils -i https://pypi.tuna.tsinghua.edu.cn/simple
import pymysql host='localhost' user='root' password='123456' database='test' charset='utf8' # 打开数据库连接 db = pymysql.connect(host=host,user=user,password=password,database=database,charset='utf8') cursor=db.cursor() # 使用 execute() 方法执行 SQL 查询 sql="select * from elementui" cursor.execute(sql) # fetchone() 方法获取单条数据.fetchall()获取所有元祖数据。fetchmany(size)获取指定条数数据 data = cursor.fetchall() # 关闭数据库连接 cursor.close() db.close()
import pymysql from dbutils.pooled_db import PooledDB POOL = PooledDB( # 使用链接数据库的模块 creator=pymysql, # 连接池允许的最大连接数,0和None表示不限制连接数 maxconnections=6, # 初始化时,链接池中至少创建的空闲的链接,0表示不创建 mincached=2, # 链接池中最多闲置的链接,0和None不限制 maxcached=5, # 链接池中最多共享的链接数量,0和None表示全部共享。 # 因为pymysql和MySQLdb等模块的 threadsafety都为1, # 所有值无论设置为多少,maxcached永远为0,所以永远是所有链接都共享。 maxshared=3, # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错 blocking=True, # 一个链接最多被重复使用的次数,None表示无限制 maxusage=None, # 开始会话前执行的命令列表。如:["set datestyle to ...", "set time zone ..."] setsession=[], # ping MySQL服务端,检查是否服务可用。 # 如:0 = None = never, 1 = default = whenever it is requested, # 2 = when a cursor is created, 4 = when a query is executed, 7 = always ping=0, # 主机地址 host='127.0.0.1', # 端口 port=3306, # 数据库用户名 user='root', # 数据库密码 password='123456', # 数据库名 database='test', # 字符编码 charset='utf8' ) def func(): # 检测当前正在运行连接数的是否小于最大链接数,如果不小于则:等待或报raise TooManyConnections异常 # 否则则优先去初始化时创建的链接中获取链接 SteadyDBConnection。 # 然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。 # 如果最开始创建的链接没有链接,则去创建一个SteadyDBConnection对象,再封装到PooledDedicatedDBConnection中并返回。 # 一旦关闭链接后,连接就返回到连接池让后续线程继续使用。 # 创建连接,POOL数据库连接池中 conn = POOL.connection() # 创建游标 cursor = conn.cursor() # SQL语句 cursor.execute('select * from tb1') # 执行结果 result = cursor.fetchall() # 将conn释放,放回连接池 conn.close()
1、封装连接池
''' 连接池初始化模块,做了什么 1、设置类变量 pool 池, 2、设置方法 getconn ,作用通过 dbutils 模块的 PooledDB 方法连接数据库,给 pool 赋值 3、设置方法 getMyConn,返回自身实例对象 核心作用:创建一个类,此类可以创建一个连接池 pool ''' from dbutils.pooled_db import PooledDB import pymysql # import db_config as config """ @功能:创建数据库连接池 """ # 数据库信息 DB_TEST_HOST = "127.0.0.1" DB_TEST_PORT = 3306 DB_TEST_USER = "root" DB_TEST_PASSWORD = "123456" DB_TEST_DBNAME = "test" # 数据库连接编码 DB_CHARSET = "utf8" # mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接) DB_MIN_CACHED = 10 # maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小) DB_MAX_CACHED = 10 # maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用 DB_MAX_SHARED = 20 # maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制) DB_MAX_CONNECYIONS = 100 # blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......> 其他代表阻塞直到连接数减少,连接被分配) DB_BLOCKING = True # maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开) DB_MAX_USAGE = 0 # setsession : 一个可选的SQL命令列表用于准备每个会话,如["set datestyle to german", ...] DB_SET_SESSION = None # creator : 使用连接数据库的模块 DB_CREATOR = pymysql class MyConnectionPool(object): __pool = None # def __init__(self): # self.conn = self.__getConn() # self.cursor = self.conn.cursor() # 创建数据库连接conn和游标cursor def __enter__(self): self.conn = self.__getconn() self.cursor = self.conn.cursor() # 创建数据库连接池 def __getconn(self): if self.__pool is None: self.__pool = PooledDB( creator=DB_CREATOR, mincached=DB_MIN_CACHED, maxcached=DB_MAX_CACHED, maxshared=DB_MAX_SHARED, maxconnections=DB_MAX_CONNECYIONS, blocking=DB_BLOCKING, maxusage=DB_MAX_USAGE, setsession=DB_SET_SESSION, host=DB_TEST_HOST, port=DB_TEST_PORT, user=DB_TEST_USER, passwd=DB_TEST_PASSWORD, db=DB_TEST_DBNAME, use_unicode=True, charset=DB_CHARSET ) return self.__pool.connection() # 释放连接池资源 def __exit__(self, exc_type, exc_val, exc_tb): self.cursor.close() self.conn.close() # 关闭连接归还给链接池 # def close(self): # self.cursor.close() # self.conn.close() # 从连接池中取出一个连接 def getconn(self): conn = self.__getconn() cursor = conn.cursor() return cursor, conn # 获取连接池,实例化 def get_my_connection(): return MyConnectionPool()
2、封装常用查询语句 select insert delete update
from db_dbutils_init import get_my_connection """执行语句查询有结果返回结果没有返回0;增/删/改返回变更数据条数,没有返回0""" class JDBCUtills(object): def __init__(self): self.db = get_my_connection() # 从数据池中获取连接 def __new__(cls, *args, **kwargs): if not hasattr(cls, 'inst'): # 单例 cls.inst = super(JDBCUtills, cls).__new__(cls, *args, **kwargs) return cls.inst # 封装执行命令 def execute(self, sql, param=None, autoclose=False): """ 【主要判断是否有参数和是否执行完就释放连接】 :param sql: 字符串类型,sql语句 :param param: sql语句中要替换的参数"select %s from tab where id=%s" 其中的%s就是参数 :param autoclose: 是否关闭连接 :return: 返回连接conn和游标cursor """ cursor, conn = self.db.getconn() # 从连接池获取连接 count = 0 try: # count : 为改变的数据条数 if param: count = cursor.execute(sql, param) else: count = cursor.execute(sql) conn.commit() if autoclose: self.close(cursor, conn) except Exception as e: pass return cursor, conn, count # 释放连接 def close(self, cursor, conn): """释放连接归还给连接池""" cursor.close() conn.close() # 查询所有 def selectall(self, sql, param=None): try: cursor, conn, count = self.execute(sql, param) res = cursor.fetchall() return res except Exception as e: print(e) self.close(cursor, conn) return count # 查询单条 def selectone(self, sql, param=None): try: cursor, conn, count = self.execute(sql, param) res = cursor.fetchone() self.close(cursor, conn) return res except Exception as e: print("error_msg:", e.args) self.close(cursor, conn) return count # 增加 def insertone(self, sql, param): try: cursor, conn, count = self.execute(sql, param) # _id = cursor.lastrowid() # 获取当前插入数据的主键id,该id应该为自动生成为好 conn.commit() self.close(cursor, conn) return count # 防止表中没有id返回0 # if _id == 0: # return True # return _id except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count # 增加多行 def insertmany(self, sql, param): """ :param sql: :param param: 必须是元组或列表[(),()]或((),()) :return: """ cursor, conn, count = self.db.getconn() try: cursor.executemany(sql, param) conn.commit() return count except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count # 删除 def delete(self, sql, param=None): try: cursor, conn, count = self.execute(sql, param) self.close(cursor, conn) return count except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count # 更新 def update(self, sql, param=None): try: cursor, conn, count = self.execute(sql, param) conn.commit() self.close(cursor, conn) return count except Exception as e: print(e) conn.rollback() self.close(cursor, conn) return count
from jdbcUtills import JDBCUtills def test1(): db=JDBCUtills() sql='select * from elementui where id=%s' args=1 res=db.selectone(sql,args) print(res)
以上是脚本宝典为你收集整理的使用python连接mysql数据库全部内容,希望文章能够帮你解决使用python连接mysql数据库所遇到的问题。
本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ:384754419,请注明来意。