Python操做MySQL數據庫(二)

pymsql是Python中操做MySQL的模塊,其使用方法和MySQLdb幾乎相同。

下載安裝:python

pip install pymysql

1.執行SQL語句

#!/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()

2.獲取新建立數據自增ID

#!/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

3.獲取查詢數據

#!/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

  • cursor.scroll(1,mode='relative')  # 相對當前位置移動
  • cursor.scroll(2,mode='absolute') # 相對絕對位置移動

4.fetch數據類型默認返回的是元祖類型,返回字典類型以下:

#!/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()

5.也能夠將其封裝成類,包含增刪改查方法,使用時直接導入調用便可

安裝數據庫鏈接池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
相關文章
相關標籤/搜索