Python接入MySQL与PostgreSQL数据库完整指南

Python作为后端开发的主流语言，与关系型数据库的集成是构建现代应用的基础。本文将从核心流程到优化方案，用代码+最佳实践的方式，快速带你上手MySQL和PostgreSQL的Python接入。

1. 环境准备与依赖安装

1.1 选择合适的驱动与工具

以下是经过筛选的轻量/实用依赖：

# MySQL驱动（选一个即可）
pip install PyMySQL  # 纯Python，跨平台易安装
pip install mysql-connector-python  # Oracle官方驱动

# PostgreSQL驱动
pip install psycopg2-binary  # 预编译版本，推荐

# 通用工具
pip install SQLAlchemy  # 通用ORM（可选，但强烈推荐）
pip install DBUtils  # 通用连接池（原生驱动时用）
pip install python-dotenv  # 环境变量管理（安全最佳实践）

1.2 配置环境变量（避坑第一步）

别把密码写死在代码里！用.env文件管理敏感信息：

# .env
# MySQL
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=app_user
MYSQL_PASSWORD=your_strong_password
MYSQL_DATABASE=test_db

# PostgreSQL
PG_HOST=localhost
PG_PORT=5432
PG_USER=app_user
PG_PASSWORD=your_strong_password
PG_DATABASE=test_db
# config.py
from dotenv import load_dotenv
import os

load_dotenv()

DB_CONFIG = {
    "mysql": {
        "host": os.getenv("MYSQL_HOST"),
        "port": int(os.getenv("MYSQL_PORT", 3306)),
        "user": os.getenv("MYSQL_USER"),
        "password": os.getenv("MYSQL_PASSWORD"),
        "database": os.getenv("MYSQL_DATABASE"),
        "charset": "utf8mb4",
        "autocommit": False,
    },
    "postgresql": {
        "host": os.getenv("PG_HOST"),
        "port": int(os.getenv("PG_PORT", 5432)),
        "user": os.getenv("PG_USER"),
        "password": os.getenv("PG_PASSWORD"),
        "database": os.getenv("PG_DATABASE"),
    }
}

2. 原生驱动基础操作

先看原生驱动的最简实现，了解底层逻辑后再用ORM。

2.1 通用连接上下文（避免资源泄露）

不管MySQL还是PostgreSQL，原生连接都需要自动关闭/回滚，用上下文管理器封装最方便：

import logging
from contextlib import contextmanager

logging.basicConfig(level=logging.INFO)

# MySQL上下文
def mysql_conn_context(config):
    import pymysql
    conn = None
    try:
        conn = pymysql.connect(**config)
        yield conn
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        logging.error(f"MySQL操作失败: {e}")
        raise
    finally:
        if conn:
            conn.close()

# PostgreSQL上下文
def pg_conn_context(config):
    import psycopg2
    conn = None
    try:
        conn = psycopg2.connect(**config)
        yield conn
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        logging.error(f"PostgreSQL操作失败: {e}")
        raise
    finally:
        if conn:
            conn.close()

2.2 MySQL CRUD示例

from config import DB_CONFIG

def create_mysql_user(name, email, age):
    """创建用户（带返回ID）"""
    with mysql_conn_context(DB_CONFIG["mysql"]) as conn:
        cursor = conn.cursor()
        sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
        cursor.execute(sql, (name, email, age))
        return cursor.lastrowid

def get_mysql_users(limit=10):
    """获取用户列表（返回字典格式）"""
    with mysql_conn_context(DB_CONFIG["mysql"]) as conn:
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute("SELECT * FROM users LIMIT %s", (limit,))
        return cursor.fetchall()

2.3 PostgreSQL特有CRUD小技巧

PostgreSQL支持RETURNING直接返回插入/更新后的数据，还有字典参数更易读：

from config import DB_CONFIG

def upsert_pg_user(data):
    """UPSERT操作（插入冲突则更新，PostgreSQL特有）"""
    with pg_conn_context(DB_CONFIG["postgresql"]) as conn:
        cursor = conn.cursor()
        sql = """
            INSERT INTO users (name, email, age)
            VALUES (%(name)s, %(email)s, %(age)s)
            ON CONFLICT (email)  # 指定冲突列（需有唯一索引）
            DO UPDATE SET age = EXCLUDED.age
            RETURNING id, name, email
        """
        cursor.execute(sql, data)
        return cursor.fetchone()

3. 连接池：性能提升核心

原生连接每次建立/断开都要花几十毫秒，连接池复用连接可以提升10倍以上的并发性能。

3.1 MySQL连接池（DBUtils）

from DBUtils.PooledDB import PooledDB
import pymysql

# 全局初始化连接池
mysql_pool = PooledDB(
    creator=pymysql,
    maxconnections=20,  # 最大连接数（根据机器内存调整）
    mincached=2,        # 初始空闲连接
    maxcached=5,        # 最大空闲连接
    **DB_CONFIG["mysql"]
)

# 替换原来的上下文
def mysql_pool_context():
    conn = None
    try:
        conn = mysql_pool.connection()
        yield conn
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        logging.error(f"MySQL连接池操作失败: {e}")
        raise
    finally:
        if conn:
            conn.close()  # 这里的close是归还到池，不是真断开

3.2 PostgreSQL连接池（SQLAlchemy自带更简单）

如果已经用SQLAlchemy，它的连接池开箱即用：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# PostgreSQL连接池（默认已启用）
pg_engine = create_engine(
    "postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}".format(**DB_CONFIG["postgresql"]),
    pool_size=10,
    max_overflow=10,  # 超出pool_size后的临时连接数
    pool_pre_ping=True  # 自动检测断开的连接
)
PgSession = sessionmaker(bind=pg_engine)

4. SQLAlchemy ORM：简化开发的利器

ORM（对象关系映射）把数据库表映射成Python类，让你写SQL像写代码一样，还能自动处理参数化、跨库兼容。

4.1 定义模型

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True, nullable=False, index=True)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

# 创建表（生产环境用Alembic迁移）
Base.metadata.create_all(bind=pg_engine)

4.2 ORM CRUD

from contextlib import contextmanager

@contextmanager
def pg_orm_session():
    session = PgSession()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        logging.error(f"ORM操作失败: {e}")
        raise
    finally:
        session.close()

# 创建用户
def create_orm_user(name, email, age):
    with pg_orm_session() as session:
        user = User(name=name, email=email, age=age)
        session.add(user)
        session.flush()  # 不用提交就能获取ID
        return user.id

# 查询用户
def get_young_users(min_age=18, max_age=30, limit=10):
    with pg_orm_session() as session:
        return session.query(User)\
                     .filter(User.age.between(min_age, max_age))\
                     .order_by(User.created_at.desc())\
                     .limit(limit)\
                     .all()

5. 避坑与最佳实践

5.1 SQL注入防护（重中之重）

永远不要拼接用户输入到SQL！原生驱动用%s/%s（PostgreSQL用%s或%(key)s），ORM自动处理。

✅ 正确示例（原生）：

cursor.execute("SELECT * FROM users WHERE name = %s", (user_input_name,))

❌ 错误示例（拼接）：

cursor.execute(f"SELECT * FROM users WHERE name = '{user_input_name}'")

5.2 事务管理

默认情况下，原生驱动和SQLAlchemy的autocommit都是False，这是安全的——只有显式调用commit()才会保存数据，出错时调用rollback()回滚。

5.3 索引设计

索引能大幅提升查询速度，但也会降低写入速度，设计原则：

  1. 给经常用在WHERE/JOIN/ORDER BY的字段加索引（如示例中的email）
  2. 避免给小表/更新频繁的字段加索引
  3. PostgreSQL支持CONCURRENTLY在线加索引，不会锁表
1. 轻量项目用**原生驱动+连接池** 2. 复杂项目/需要跨库兼容用**SQLAlchemy ORM** 3. 生产环境必须用**环境变量管理敏感信息**和**连接池**

相关教程

总结

Python接入MySQL和PostgreSQL的核心逻辑一致，区别主要在驱动语法和PostgreSQL的特有功能（如UPSERT、JSONB）。记住连接池、参数化查询、事务管理这三个关键点，就能写出高效、安全的数据库代码。