本文目录导读：
我来详细介绍Python操作数据库的各种方案,从基础到进阶：
Python数据库操作基础方案
MySQL
# 安装：pip install pymysql
import pymysql
# 连接数据库
conn = pymysql.connect(
host='localhost',
user='root',
password='password',
database='test_db',
charset='utf8mb4'
)
cursor = conn.cursor()
# 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100),
age INT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 插入数据
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
cursor.execute(sql, ('张三', 25))
conn.commit()
# 查询数据
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
for row in results:
print(row)
cursor.close()
conn.close()
PostgreSQL
# 安装：pip install psycopg2-binary
import psycopg2
conn = psycopg2.connect(
host='localhost',
database='test_db',
user='postgres',
password='password'
)
cursor = conn.cursor()
# 创建表和操作类似MySQL
cursor.close()
conn.close()
ORM框架方案（推荐）
SQLAlchemy（最流行）
# 安装：pip install sqlalchemy pymysql
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
Base = declarative_base()
# 定义模型
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(100))
age = Column(Integer)
created_at = Column(DateTime, default=datetime.now)
# 创建连接
engine = create_engine('mysql+pymysql://root:password@localhost/test_db')
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
# 增删改查
# 添加
new_user = User(name='李四', age=30)
session.add(new_user)
session.commit()
# 查询
users = session.query(User).filter(User.age > 20).all()
for user in users:
print(user.name, user.age)
# 更新
user = session.query(User).filter_by(name='李四').first()
user.age = 31
session.commit()
# 删除
session.delete(user)
session.commit()
session.close()
Django ORM（配合Django框架）
# settings.py 配置
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.mysql',
'NAME': 'test_db',
'USER': 'root',
'PASSWORD': 'password',
'HOST': 'localhost',
'PORT': '3306',
}
}
# models.py
from django.db import models
class User(models.Model):
name = models.CharField(max_length=100)
age = models.IntegerField()
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
db_table = 'users'
# 使用
from myapp.models import User
# 创建
User.objects.create(name='王五', age=28)
# 查询
users = User.objects.filter(age__gt=20)
# 更新
User.objects.filter(name='王五').update(age=29)
# 删除
User.objects.filter(name='王五').delete()
连接池方案（高性能）
# 安装：pip install DBUtils
from dbutils.pooled_db import PooledDB
import pymysql
# 创建连接池
pool = PooledDB(
creator=pymysql,
maxconnections=6, # 最大连接数
mincached=2, # 初始化时创建的连接数
maxcached=5, # 最大空闲连接数
host='localhost',
user='root',
password='password',
database='test_db',
charset='utf8mb4'
)
# 从池中获取连接
conn = pool.connection()
cursor = conn.cursor()
# 执行SQL
cursor.execute("SELECT * FROM users")
result = cursor.fetchall()
# 归还连接到池
cursor.close()
conn.close()
异步数据库操作（高性能场景）
# 安装：pip install aiomysql
import asyncio
import aiomysql
async def async_db_operations():
# 创建连接池
pool = await aiomysql.create_pool(
host='localhost',
port=3306,
user='root',
password='password',
db='test_db',
charset='utf8mb4',
autocommit=True
)
async with pool.acquire() as conn:
async with conn.cursor() as cur:
# 执行查询
await cur.execute("SELECT * FROM users")
result = await cur.fetchall()
print(result)
# 插入数据
await cur.execute(
"INSERT INTO users (name, age) VALUES (%s, %s)",
('AsyncUser', 22)
)
pool.close()
await pool.wait_closed()
# 运行异步操作
asyncio.run(async_db_operations())
数据库迁移工具
Alembic（配合SQLAlchemy）
# 安装：pip install alembic # 初始化：alembic init alembic # alembic/env.py 配置 from alembic import context from sqlalchemy import engine_from_config from myapp.models import Base # 导入你的模型 target_metadata = Base.metadata # 创建迁移：alembic revision --autogenerate -m "add_users" # 执行迁移：alembic upgrade head
实用工具类封装
class DatabaseManager:
def __init__(self, config):
self.config = config
self.conn = None
self.cursor = None
def __enter__(self):
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
def connect(self):
self.conn = pymysql.connect(**self.config)
self.cursor = self.conn.cursor()
def execute(self, sql, params=None):
try:
if params:
self.cursor.execute(sql, params)
else:
self.cursor.execute(sql)
self.conn.commit()
return self.cursor
except Exception as e:
self.conn.rollback()
raise e
def fetch_all(self, sql, params=None):
self.execute(sql, params)
return self.cursor.fetchall()
def close(self):
if self.cursor:
self.cursor.close()
if self.conn:
self.conn.close()
# 使用
config = {
'host': 'localhost',
'user': 'root',
'password': 'password',
'database': 'test_db'
}
with DatabaseManager(config) as db:
users = db.fetch_all("SELECT * FROM users WHERE age > %s", (20,))
print(users)
最佳实践建议
配置管理
# config.py
DATABASE_CONFIG = {
'mysql': {
'host': os.getenv('DB_HOST', 'localhost'),
'user': os.getenv('DB_USER', 'root'),
'password': os.getenv('DB_PASSWORD', ''),
'database': os.getenv('DB_NAME', 'test'),
'port': int(os.getenv('DB_PORT', 3306)),
'charset': 'utf8mb4',
'autocommit': True
}
}
错误处理和日志
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def safe_query(sql, params=None):
try:
with DatabaseManager(config) as db:
return db.fetch_all(sql, params)
except Exception as e:
logger.error(f"Database error: {e}")
raise
选择合适的方案：
- 简单项目：直接使用pymysql/psycopg2
- 中型项目：SQLAlchemy ORM
- Django项目：Django ORM
- 高性能要求：异步库 + 连接池
- 需要迁移：Alembic + SQLAlchemy
需要我详细讲解某个特定方案的具体实现吗？
