pymysql操作mysql數據庫

1.建庫

import pymysql
# 建庫
try:
    conn=pymysql.connect(
        host='127.0.0.1',
        port=3306,
        user='root',
        passwd='123456',
    )
    cur=conn.cursor()
    create_database_sql='CREATE DATABASE IF NOT EXISTS py3_tstgr DEFAULT CHARSET utf8 COLLATE utf8_general_ci;'
    cur.execute(create_database_sql)
    cur.close()
    print('創建數據庫 py3_tstgr 成功！')
except pymysql.Error as e:
    print('pymysql.Error: ',e.args[0],e.args[1])

 

2.建表

import pymysql
# 建表
try:
    conn=pymysql.connect(
        host='127.0.0.1',
        port=3306,
        user='root',
        passwd='123456',
        db='py3_tstgr',
        charset='utf8'
    )
    cur=conn.cursor()
    cur.execute('drop table if exists user;')
    create_table_sql='''
        CREATE TABLE user(
            id int(11) DEFAULT NULL ,
            name VARCHAR(50) DEFAULT NULL ,
            password VARCHAR(30) DEFAULT NULL ,
            birthday TIMESTAMP DEFAULT now() 
        )engine=innodb DEFAULT CHARACTER set utf8;
    '''
    cur.execute(create_table_sql)
    print('創建數據庫表成功！')
except pymysql.Error as e:
    print('mysql.Error: ',e.args[0],e.args[1]) 

 

3.增、改、刪、查數據操作

import pymysql
# 數據庫操作
# 連接
def db_conn(host,port,user,passwd,db_name):
    try:
        conn=pymysql.connect(
            host=host,
            port=port,
            user=user',
            passwd=password,
            db=db_name,
            charset='utf8',
        )
        # cur=conn.cursor()
        print('數據庫連接成功！')
        # 返回連接
        return conn
    except pymysql.Error as e:
        print('數據庫連接失敗')
        print('mysql.Error: ',e.args[0],e.args[1])

def db_cur(conn):
    # 獲取游標
    cur=conn.cursor()
    return cur

def db_close(cur,conn):
    # 游標關閉
    cur.close()
    # 提交事務
    conn.commit()
    # 連接關閉
    conn.close()

# 插入單行數據
def db_insert_data(sql,cur,*args):
    try:
        # print(args)
        result=cur.execute(sql,args)
        print('添加語句受影響的行數：',result)
    except Exception as e:
        print('db_insert_data error: ',e.args)

# 批量插入數據
def db_insert_datas(sql,cur,list_datas):
    try:
        result=cur.executemany(sql,list_datas)
        print('批量插入受影響的行數：',result)
    except Exception as e:
        print('db_insert_datas error: ',e.args)

# 修改單行數據
def db_update(sql,cur):
    result=cur.execute(sql)
    print('修改語句受影響的行數：',result)

# 批量修改數據
def db_update_datas(sql,cur,list_datas):
    try:
        result=cur.executemany(sql,list_datas)
        print('批量修改受影響的行數：',result)
    except Exception as e:
        print('db_update_datas error: ',e.args)

# 刪除單行數據
def db_delete_data(sql,cur):
    result=cur.execute(sql)
    print('刪除語句受影響的行數：',result)

# 批量刪除數據
def db_delete_datas(sql,cur,list_datas):
    try:
        result=cur.executemany(sql,list_datas)
        print('批量刪除受影響的行數：',result)
    except Exception as e:
        print('db_delete_datas error: ',e.args)

# 回滾
def roll_back(conn):
    try:
        conn.rollback()
        print('回滾完成！')
    except Exception as e:
        print('rollback error: ',e.args)

def db_select_data(sql,cur):
    result=cur.execute(sql)
    print('查詢語句受影響的行數：',result)

def db_select_datas(sql,cur,list_datas):
    try:
        result=cur.executemany(sql,list_datas)
        print('批量查詢受影響的行數：',result)
    except Exception as e:
        print('db_select_datas error: ',e.args)

if __name__=="__main__":

    host='127.0.0.1'
    port=3306
    user='root'
    passwd='123456'
    db='py3_tstgr'
    conn=db_conn(host,port,user,passwd,db)
    print(conn)
    cur=db_cur(conn)
    print(cur)
    
    insert_sql="insert into user values(1,'tom','123','1990-01-01');"
    db_insert_data(insert_sql,cur)

    insert_sql2="insert into user values(%s,%s,%s,%s);"
    db_insert_data(insert_sql2,cur,2,'lucy','aaa','1991-02-02')
    
    insert_datas_sql="insert into user values(%s,%s,%s,%s);"
    list_datas=[(2,'ha2','222','1992-02-02'),(3,'ha3','333','1993-03-03'),(4,'ha4','444','1994-04-04')]
    db_insert_datas(insert_datas_sql,cur,list_datas)
    
    # 查詢數據
    sql="select * from user;"
    cur.execute(sql)
    res1=cur.fetchone()   # 獲取一行數據
    print(res1)
    res2=cur.fetchmany(2)  #獲取多行數據
    print(res2)
    res3=cur.fetchall()  # 獲取所有數據
    print(res3)
    print(cur.fetchone())  # None
    # 重置游標
    cur.scroll(1,mode='relative')
    print(cur.fetchone())
    
    # 更新數據
    up_sql="update user set name='lala' where password='123';"
    db_update(up_sql,cur)
    
    # 批量更新數據
    up_sqls="update user set name=%s where password=%s;"
    up_datas=[('lw','123'),('lc','aaa')]
    db_update_datas(up_sqls,cur,up_datas)
    
    # 刪除數據
    delete_sql="delete from user where name='lw';"
    db_delete_data(delete_sql,cur)
    
    delete_sqls="delete from user where name=%s;"
    db_delete_datas(delete_sqls,cur,[('lc'),('hah2'),('hah3')])

    # print(cur.rownumber)  # 0 獲取所在行號

    # cur.scroll(0,mode='absolute')
    # cur.execute('select * from user;')

    select_sql="select * from user;"
    db_select_data(select_sql,cur)
    print(cur.fetchall())
    

    # roll_back(conn)  # 回滾
    # print(cur.rownumber)
    # cur.scroll(0, mode='absolute')
    # cur.execute('select * from user;')

    # 查詢單行
    select_sql = "select * from user;"
    db_select_data(select_sql, cur)
    
    # 批量查詢
    select_sqls="select * from user where name=%s;"
    db_select_datas(select_sqls,cur,[('ha1'),('ha2'),('ha3')])
    print(cur.fetchall())

    db_close(cur,conn)  # 斷開連接