下載安裝:python
pip install pymysql
#!/usr/bin/env python # -*- coding:utf-8 -*- import pymysql # 建立鏈接 conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1') # 建立遊標 cursor = conn.cursor() # 執行SQL,並返回收影響行數 effect_row = cursor.execute("update hosts set host = '1.1.1.2'") # 執行SQL,並返回受影響行數 #effect_row = cursor.execute("update hosts set host = '1.1.1.2' where nid > %s", (1,)) # 執行SQL,並返回受影響行數 #effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)]) # 提交,否則沒法保存新建或者修改的數據 conn.commit() # 關閉遊標 cursor.close() # 關閉鏈接 conn.close()
#!/usr/bin/env python # -*- coding:utf-8 -*- import pymysql conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1') cursor = conn.cursor() cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)]) conn.commit() cursor.close() conn.close() # 獲取最新自增ID new_id = cursor.lastrowid
#!/usr/bin/env python # -*- coding:utf-8 -*- import pymysql conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1') cursor = conn.cursor() cursor.execute("select * from hosts") # 獲取第一行數據 row_1 = cursor.fetchone() # 獲取前n行數據 # row_2 = cursor.fetchmany(3) # 獲取全部數據 # row_3 = cursor.fetchall() conn.commit() cursor.close() conn.close()
注:在fetch數據時按照順序進行,可使用cursor.scroll(num,mode)來移動遊標位置,如:mysql
#!/usr/bin/env python # -*- coding:utf-8 -*- import pymysql conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1') # 遊標設置爲字典類型 cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) r = cursor.execute("call p1()") result = cursor.fetchone() conn.commit() cursor.close() conn.close()
# -*- coding:utf-8 -*- # /user/bin/python # @Time : 2018/5/23 23:32 # @Author : liyongle # @File : sql_ method.py
import contextlib import os import pymysql @contextlib.contextmanager def __get_mysql_db__( host="10.1.1.113", port=3306, user="liyongle", passwd="xxxx", db="test_lib", charset="utf8" ): con = pymysql.connect( host=host, port=port, user=user, passwd=passwd, db=db, charset=charset, cursorclass=pymysql.cursors.DictCursor, ) cur = con.cursor() try: yield cur finally: con.commit() cur.close() con.close() def run_query(query): # 這個函數返回一個列表,列表裏面的元素都是字典 with __get_mysql_db__() as cursor: cursor.execute(query) return cursor.fetchall() def main(): sql = "SELECT case_id FROM testcase LEFT JOIN req_method ON req_method.reuqest_id = testcase.method_id" run_query(sql) if __name__ == '__main__': main()
安裝數據庫鏈接池sql
pip install DBUtils
具體代碼以下:數據庫
# -*- coding:utf-8 -*- # /user/bin/python # @Time : 2018/5/24 0:16 # @Author : liyongle # @File : sql_ method.py import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection POOL = PooledDB( creator=pymysql, # 使用連接數據庫的模塊 maxconnections=20, # 鏈接池容許的最大鏈接數,0和None表示不限制鏈接數 mincached=2, # 初始化時,連接池中至少建立的空閒的連接,0表示不建立 maxcached=5, # 連接池中最多閒置的連接,0和None不限制 #maxshared=3, # 連接池中最多共享的連接數量,0和None表示所有共享。PS: 無用,由於pymysql和MySQLdb等模塊的 threadsafety都爲1,全部值不管設置爲多少,_maxcached永遠爲0,因此永遠是全部連接都共享。 blocking=True, # 鏈接池中若是沒有可用鏈接後,是否阻塞等待。True,等待;False,不等待而後報錯 maxusage=None, # 一個連接最多被重複使用的次數,None表示無限制 setsession=[], # 開始會話前執行的命令列表。如:["set datestyle to ...", "set time zone ..."] ping=0, # 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 host='192.168.11.38', port=3306, user='root', passwd='xxxxxx', db='test_db', charset='utf8' ) def connect(): # 建立鏈接 conn = POOL.connection() # 建立遊標 cursor = conn.cursor(pymysql.cursors.DictCursor) return conn,cursor def close(conn,cursor): # 關閉遊標 cursor.close() # 關閉鏈接 conn.close() def fetch_one(sql,args): conn,cursor = connect() # 執行SQL,並返回收影響行數 effect_row = cursor.execute(sql,args) result = cursor.fetchone() close(conn,cursor) return result def fetch_all(sql,args): conn, cursor = connect() # 執行SQL,並返回收影響行數 cursor.execute(sql,args) result = cursor.fetchall() close(conn, cursor) return result def insert(sql,args): """ 建立數據 :param sql: 含有佔位符的SQL :return: """ conn, cursor = connect() # 執行SQL,並返回收影響行數 effect_row = cursor.execute(sql,args) conn.commit() close(conn, cursor) def delete(sql,args): """ 建立數據 :param sql: 含有佔位符的SQL :return: """ conn, cursor = connect() # 執行SQL,並返回收影響行數 effect_row = cursor.execute(sql,args) conn.commit() close(conn, cursor) return effect_row def update(sql,args): conn, cursor = connect() # 執行SQL,並返回收影響行數 effect_row = cursor.execute(sql, args) conn.commit() close(conn, cursor) return effect_row