一、MySQL 数据库
1. 安装驱动
bash
pip install mysql-connector-python
2. 连接示例
python
import mysql.connector
from mysql.connector import Error
try:
# 建立连接
connection = mysql.connector.connect(
host='localhost',
user='your_username',
password='your_password',
database='your_database'
)
if connection.is_connected():
cursor = connection.cursor()
# 执行查询
cursor.execute("SELECT VERSION()")
version = cursor.fetchone()
print(f"数据库版本: {version}")
# 执行插入
insert_query = "INSERT INTO users (name, age) VALUES (%s, %s)"
values = ("John", 30)
cursor.execute(insert_query, values)
connection.commit() # 提交事务
print(f"插入成功，影响行数: {cursor.rowcount}")
except Error as e:
print(f"数据库错误: {e}")
finally:
if connection.is_connected():
cursor.close()
connection.close()
print("数据库连接已关闭")
二、PostgreSQL 数据库
1. 安装驱动
bash
pip install psycopg2-binary
2. 连接示例
python
import psycopg2
from psycopg2 import OperationalError
try:
# 建立连接
connection = psycopg2.connect(
host="localhost",
database="your_database",
user="your_username",
password="your_password",
port="5432"
)
cursor = connection.cursor()
# 执行查询
cursor.execute("SELECT version();")
record = cursor.fetchone()
print(f"数据库版本: {record}")
except OperationalError as e:
print(f"数据库错误: {e}")
finally:
if connection:
cursor.close()
connection.close()
print("数据库连接已关闭")
三、SQLite 数据库
1. 安装驱动
bash
# SQLite驱动内置在Python中，无需额外安装
2. 连接示例
python
import sqlite3
from sqlite3 import Error
try:
# 建立连接（如果数据库不存在，会自动创建）
connection = sqlite3.connect('example.db')
cursor = connection.cursor()
# 创建表
cursor.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY, name TEXT, age INTEGER)''')
# 插入数据
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ('Alice', 25))
connection.commit()
# 查询数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
print(row)
except Error as e:
print(f"数据库错误: {e}")
finally:
if connection:
connection.close()
四、SQLAlchemy（通用 ORM）
1. 安装库
bash
pip install sqlalchemy
2. 使用示例（以 MySQL 为例）
python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 创建引擎
engine = create_engine('mysql+mysqlconnector://user:password@localhost/mydatabase')
# 定义模型
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
# 创建表
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
# 添加数据
new_user = User(name='Bob', age=35)
session.add(new_user)
session.commit()
# 查询数据
users = session.query(User).all()
for user in users:
print(f"ID: {user.id}, Name: {user.name}, Age: {user.age}")
# 关闭会话
session.close()
五、数据库连接池
1. 安装库
bash
pip install SQLAlchemy # 自带连接池功能
# 或
pip install DBUtils # 通用连接池库
2. 使用示例（SQLAlchemy 连接池）
python
from sqlalchemy import create_engine
# 创建带连接池的引擎
engine = create_engine(
'mysql+mysqlconnector://user:password@localhost/mydatabase',
pool_size=5, # 连接池大小
max_overflow=10, # 最大溢出连接数
pool_timeout=30, # 连接超时时间（秒）
pool_recycle=3600 # 连接回收时间（秒）
)
# 使用with语句自动管理连接
with engine.connect() as connection:
result = connection.execute("SELECT * FROM users")
for row in result:
print(row)
六、常见操作与最佳实践
1. 参数化查询
python
# 防止SQL注入
query = "SELECT * FROM users WHERE age > %s" # MySQL
# query = "SELECT * FROM users WHERE age > ?" # SQLite
cursor.execute(query, (25,))
2. 事务处理
python
try:
with connection.begin():
# 执行多个操作
cursor.execute("INSERT INTO ...")
cursor.execute("UPDATE ...")
except:
# 自动回滚
print("事务失败，已回滚")
3. 错误处理
python
try:
cursor.execute(query)
except DatabaseError as e:
print(f"操作失败: {e}")
connection.rollback()
