pip install mysql-connector-python

import mysql.connector
from mysql.connector import Error

try:
    # 連接到MySQL數(shù)據(jù)庫(kù)
    connection = mysql.connector.connect(
        host='localhost',       # 數(shù)據(jù)庫(kù)服務(wù)器地址
        user='your_username',   # 數(shù)據(jù)庫(kù)用戶名
        password='your_password', # 數(shù)據(jù)庫(kù)密碼
        database='your_database_name'  # 數(shù)據(jù)庫(kù)名
    )

    if connection.is_connected():
        print("Connected to MySQL database")

        # 創(chuàng)建一個(gè)cursor對(duì)象
        cursor = connection.cursor()

        # 創(chuàng)建表
        create_table_query = """
        CREATE TABLE IF NOT EXISTS employees (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(255) NOT NULL,
            salary INT,
            department VARCHAR(255)
        );
        """
        cursor.execute(create_table_query)
        print("Table created or already exists")

        # 插入數(shù)據(jù)
        insert_query = """
        INSERT INTO employees (name, salary, department)
        VALUES (%s, %s, %s);
        """
        data = ("John Doe", 70000, "Finance")
        cursor.execute(insert_query, data)
        connection.commit()
        print("Data inserted successfully")

except Error as e:
    print("Error while connecting to MySQL", e)

finally:
    # 關(guān)閉cursor和連接
    if connection.is_connected():
        cursor.close()
        connection.close()
        print("MySQL connection is closed")

#encoding=gbk
import  sys
import  MySQLdb
 
 
conn=MySQLdb.Connection('127.0.0.1','root','123456','job',charset='gbk')
cur=conn.cursor()
cur.execute("select * from demo_jobs_store")
conn.commit()
conn.close()
 
cur.scroll(0)
row1=cur.fetchone()
print row1[3]

import pymysql
 
## 登錄賬號(hào)
host = "host_name"
user = "user_name"
password = "password"
database = "database_name"
port = port
 
conn = pymysql.connect(host, user, passwd, db, port) ## 登錄數(shù)據(jù)庫(kù)，返回連接
 
cursor = conn.cursor()
sql = "select var1, ..., varn from table_name where ..."
cursor.execute(sql)
 
data = cursor.fetchall() ##data為以元組為元素的元組
conn.commit()
cursor.close()
conn.close() ## 關(guān)閉連接

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/18 15:20
# @Function:
import pymysql
 
 
class my_db:
    '''
    動(dòng)作類：獲取數(shù)據(jù)連接，連接ip,端口，賬號(hào)密碼。。
    '''
 
    # 構(gòu)造函數(shù)
    def __init__(self):
        try:
            self.dbconn = pymysql.connect(host='127.0.0.1', port=3306, user='root', password='root',
                                          database='test_cases', charset='utf8')
        except Exception as e:
            print('初始化數(shù)據(jù)庫(kù)連接失?。?s' % e)
 
    def close(self):
        self.dbconn.close()
 
    # query=查詢語(yǔ)句
    def select_record(self, query):
        # 查詢數(shù)據(jù)
        print('query:%s' % query)
        try:
            # 建立游標(biāo)
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            result = db_cursor.fetchall()
            return result
        except Exception as e:
            print('數(shù)據(jù)庫(kù)查詢數(shù)據(jù)失?。?s' % e)
            db_cursor.close()
            exit()
 
    # 插入
    def execute_insert(self, query):
        print('query:%s' % query)
        try:
            # 建立游標(biāo)
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            db_cursor.execute('commit')
            return True
        except Exception as e:
            print('數(shù)據(jù)庫(kù)插入數(shù)據(jù)失?。?s' % e)
            # 事務(wù)回滾
            db_cursor.execute('rollback')
            db_cursor.close()
            exit()

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/7 15:07
# @Function:
# 針對(duì)數(shù)據(jù)庫(kù)中的數(shù)據(jù)做數(shù)據(jù)驅(qū)動(dòng)
import time
import unittest
from ddt import ddt, data, unpack
 
from key_demo.tool.MyDB import my_db
 
# 必須要實(shí)例化，不然會(huì)報(bào)TypeError: select_record() missing 1 required positional argument: 'query'
testdb = my_db()
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
 
 
@ddt
class case(unittest.TestCase):
    # @data(*my_db().select_record('select weaid,success from weather'))
    @data(*testdb.select_record('select weaid,success from weather'))
    @unpack
    def test_1(self, weaid, success):
        print(weaid, success)
        testdb.execute_insert('insert into weather(weaid,success,cre_time) values ("2","2","%s")' % now)
 
 
if __name__ == '__main__':
    unittest.main()

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/7 15:07
# @Function:
# 針對(duì)數(shù)據(jù)庫(kù)中的數(shù)據(jù)做數(shù)據(jù)驅(qū)動(dòng)
import time
import unittest
from ddt import ddt, data, unpack
 
from key_demo.tool.MyDB import my_db
 
# 必須要實(shí)例化，不然會(huì)報(bào)TypeError: select_record() missing 1 required positional argument: 'query'
testdb = my_db()
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
 
 
@ddt
class case(unittest.TestCase):
    # @data(*my_db().select_record('select weaid,success from weather'))
    @data(testdb.execute_insert('insert into weather(weaid,success,cre_time)values("2","2","%s")' % now))
    # @unpack
    def test_1(self,t):
        print(t)
        res = testdb.select_record('select weaid,success from weather')
        print(res)
 
 
if __name__ == '__main__':
    unittest.main()