Python脚本如何操作数据库计划

本文目录导读：

1. Python数据库操作基础方案
2. ORM框架方案（推荐）
3. 连接池方案（高性能）
4. 异步数据库操作（高性能场景）
5. 数据库迁移工具
6. 实用工具类封装
7. 最佳实践建议

我来详细介绍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

需要我详细讲解某个特定方案的具体实现吗？