一、 Pip換源
可以查看文章路徑
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple numpy
二、 解耦
1.解耦思想
具體查看文章python解耦重構,提高程序維護性
https://editor.csdn.net/md/?articleId=140161169
- mysql 連接解耦
主要實現方式為mysql配置項的提取,其中mysql 安裝方式查看mysql8.0.19安裝zip版本
https://editor.csdn.net/md/?articleId=140093551。
三、解耦實現
- Pip換源安裝庫
pip install mysql-connector-python
pip install python-dotenv
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple mysql-connector-python
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple python-dotenv
- 配置文件 env 版本
配置文件填寫,a.env。
具體參數 DB_HOST ip , DB_DATABASE DATABASE 名稱 , DB_USER 用戶名 ,DB_PASSWORD 密碼。
DB_HOST=127.0.0.1
DB_DATABASE=customer_feedback
DB_USER=root
DB_PASSWORD=*******
- 配置文件 .ini 版本
如果想用.ini結尾的可以查看,文章python解耦重構,提高程序維護性。
4 .python 代碼實現
解耦前
connection = mysql.connector.connect(host='your_host', # 數據庫主機地址database='your_database', # 數據庫名稱user='your_user', # 數據庫用戶名password='your_password' # 數據庫密碼)
解耦后
db_host = os.getenv('DB_HOST')db_database = os.getenv('DB_DATABASE')db_user = os.getenv('DB_USER')db_password = os.getenv('DB_PASSWORD')print(db_host)# 連接數據庫connection = mysql.connector.connect(host=db_host,database=db_database,user=db_user,password=db_password)
全部代碼實現(mysql 8.0版本使用)
import mysql.connector
from mysql.connector import Error
import os
from dotenv import load_dotenv
# 加載.env文件
load_dotenv('a.env',verbose=True)
def read_data_from_mysql():try:# 連接數據庫# connection = mysql.connector.connect(# host='your_host', # 數據庫主機地址# database='your_database', # 數據庫名稱# user='your_user', # 數據庫用戶名# password='your_password' # 數據庫密碼# )# 從環境變量中讀取數據庫連接信息db_host = os.getenv('DB_HOST')db_database = os.getenv('DB_DATABASE')db_user = os.getenv('DB_USER')db_password = os.getenv('DB_PASSWORD')print(db_host)# 連接數據庫connection = mysql.connector.connect(host=db_host,database=db_database,user=db_user,password=db_password)if connection.is_connected():db_Info = connection.get_server_info()print(f"Successfully connected to MySQL Server version {db_Info}")# 創建游標對象cursor = connection.cursor()# 執行查詢query = "SELECT * FROM feedback_2024" # 更換為你的表名cursor.execute(query)# 獲取所有記錄列表records = cursor.fetchall()print("Total number of rows in table: ", cursor.rowcount)print("\nPrinting each row")for row in records:print(row)except Error as e:print(f"Error reading data from MySQL: {e}")finally:if (connection.is_connected()):cursor.close()connection.close()print("MySQL connection is closed")# 調用函數
read_data_from_mysql()