前言
在python项目开发中，我们经常使用多种数据库存储数据,常用的有：
MySQL：mysql-connector-python, PyMySQL
PostgreSQL：psycopg2
SQLite：sqlite3（标准库）
下面是一个通用的步骤示例，展示如何使用Python连接数据库并执行查询。这里我们以MySQL数据库为例，使用mysql-connector-python库。
步骤一：安装库
首先，确保你已经安装了mysql-connector-python库。如果没有安装，可以使用以下命令进行安装：
pip install mysql-connector-python
步骤二：连接数据库
接下来，编写代码来连接到数据库。
import mysql.connector
from mysql.connector import Error

def create_connection(host_name, user_name, user_password, db_name):
connection = None
try:
connection = mysql.connector.connect(
host=host_name,
user=user_name,
passwd=user_password,
database=db_name
)
print(“Connection to MySQL DB successful”)
except Error as e:
print(f"The error ‘{e}’ occurred")
return connection

步骤三：执行查询
定义一个函数来执行查询：
def execute_query(connection, query):
cursor = connection.cursor()
try:
cursor.execute(query)
connection.commit()
print(“Query executed successfully”)
except Error as e:
print(f"The error ‘{e}’ occurred")

步骤四：执行读取查询
定义一个函数来读取数据：
def execute_read_query(connection, query):
cursor = connection.cursor()
result = None
try:
cursor.execute(query)
result = cursor.fetchall()
return result
except Error as e:
print(f"The error ‘{e}’ occurred")
return None

步骤五：使用示例
使用上面定义的函数连接数据库并执行查询：

数据库连接信息

host_name = “your_host”
user_name = “your_username”
user_password = “your_password”
db_name = “your_database”

连接数据库

connection = create_connection(host_name, user_name, user_password, db_name)

创建表示例

create_table_query = “”"
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name TEXT NOT NULL,
age INT,
gender TEXT,
nationality TEXT
);
“”"
execute_query(connection, create_table_query)

插入数据示例

insert_user_query = “”"
INSERT INTO users (name, age, gender, nationality) VALUES
(‘James’, 25, ‘male’, ‘USA’),
(‘Leila’, 32, ‘female’, ‘France’);
“”"
execute_query(connection, insert_user_query)

查询数据示例

select_users_query = “SELECT * FROM users”
users = execute_read_query(connection, select_users_query)

for user in users:
print(user)

下面附上完整使用代码：

import mysql.connector
from mysql.connector import Error

def create_connection(host_name, user_name, user_password, db_name):
connection = None
try:
connection = mysql.connector.connect(
host=host_name,
user=user_name,
passwd=user_password,
database=db_name
)
print(“Connection to MySQL DB successful”)
except Error as e:
print(f"The error ‘{e}’ occurred")
return connection

def execute_query(connection, query):
cursor = connection.cursor()
try:
cursor.execute(query)
connection.commit()
print(“Query executed successfully”)
except Error as e:
print(f"The error ‘{e}’ occurred")

def execute_read_query(connection, query):
cursor = connection.cursor()
result = None
try:
cursor.execute(query)
result = cursor.fetchall()
return result
except Error as e:
print(f"The error ‘{e}’ occurred")
return None

数据库连接信息

host_name = “your_host”
user_name = “your_username”
user_password = “your_password”
db_name = “your_database”

连接数据库

connection = create_connection(host_name, user_name, user_password, db_name)

创建表示例

create_table_query = “”"
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name TEXT NOT NULL,
age INT,
gender TEXT,
nationality TEXT
);
“”"
execute_query(connection, create_table_query)

插入数据示例

insert_user_query = “”"
INSERT INTO users (name, age, gender, nationality) VALUES
(‘James’, 25, ‘male’, ‘USA’),
(‘Leila’, 32, ‘female’, ‘France’);
“”"
execute_query(connection, insert_user_query)

查询数据示例

select_users_query = “SELECT * FROM users”
users = execute_read_query(connection, select_users_query)

for user in users:
print(user)

这里在为大家展示一下实际项目中常用的model实例，类似于java中的实体：

class AlertStatistics(SurrogatePK, Model):
tablename = ‘alert_statistics’

high_count = Column(db.Integer, nullable=False, default=0)
middle_count = Column(db.Integer, nullable=False, default=0)
lower_count = Column(db.Integer, nullable=False, default=0)
handled_count = Column(db.Integer, nullable=False, default=0)
unhandled_count = Column(db.Integer, nullable=False, default=0)
alert_count = Column(db.Integer, nullable=False, default=0)
data1_count = Column(db.Integer, nullable=False, default=0)
data2_count = Column(db.Integer, nullable=False, default=0)report_date = Column(db.Date, nullable=False, default=0)
created_at = Column(db.DateTime, nullable=False, default=dt.datetime.now)