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

在 Python 后端开发中，与relational-database打交道是绕不开的核心环节。无论你是刚入门的新手，还是希望梳理最佳实践的开发者，本文都会用 “从基础连接到生产优化” 的方式，带你掌握 Python 接入 MySQL 和 PostgreSQL 的关键技能。我们会配合大量可运行的代码，让你在阅读的同时就能动手实践。

1. 环境准备与依赖安装

1.1 选择驱动与工具

针对 MySQL 和 PostgreSQL，主流的驱动如下：

# 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 文件管理连接参数，配合 python-dotenv 解析，既安全又方便。

# .env 文件示例
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=app_user
MYSQL_PASSWORD=your_strong_password
MYSQL_DATABASE=test_db

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()  # 自动读取 .env 文件并注入环境变量

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"),
    }
}

这样做有两个好处：一是密码不上传 Git，二是切换环境只需修改 .env 文件。

2. 原生驱动基础操作

先用原生的 PyMySQL 和 psycopg2 走通连接、查询、插入流程，理解底层机制后再引入 ORM 会更清晰。

2.1 通用连接上下文管理器

原生连接需要手动关闭和回滚，用 Python 的上下文管理器封装，可以确保每一步都安全。

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):
    """创建用户，并返回自增主键"""
    with mysql_conn_context(DB_CONFIG["mysql"]) as conn:
        cursor = conn.cursor()
        # 注意：参数化查询必须使用 %s 占位符
        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:
        # 指定 DictCursor 让结果更容易使用
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute("SELECT * FROM users LIMIT %s", (limit,))
        return cursor.fetchall()

2.3 PostgreSQL 特有技巧：RETURNING 与字典参数

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)                -- 要求 email 列有唯一索引
            DO UPDATE SET age = EXCLUDED.age
            RETURNING id, name, email
        """
        cursor.execute(sql, data)
        return cursor.fetchone()   # 直接拿到插入/更新后的行

这里用 %(name)s 的形式，直接传入字典 data，可读性比顺序占位符好得多。

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

原生的数据库连接每次都要经过 TCP 握手、认证等流程，开销在几十毫秒量级。对于高并发场景，使用连接池复用连接，可以将性能提升 10 倍以上。

3.1 MySQL 连接池（DBUtils）

利用 DBUtils.PooledDB 为 PyMySQL 加上连接池能力：

from DBUtils.PooledDB import PooledDB
import pymysql

# 全局初始化连接池（程序启动时执行一次）
mysql_pool = PooledDB(
    creator=pymysql,           # 使用 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_pre_ping=True      # 每次取出连接前先探测是否有效
)

PgSession = sessionmaker(bind=pg_engine)

4. SQLAlchemy ORM：让数据库操作更 Pythonic

ORM（对象关系映射）把数据库表映射成 Python 类，写增删改查就像操作普通对象，同时自动处理参数化、事务和跨库兼容。

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()

对比原生 SQL，ORM 代码更接近业务逻辑，字段改动时也更容易维护。

5. 避坑与最佳实践

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

永远不要拼接用户输入到 SQL 字符串中！ 这是安全红线。

✅ 正确做法——使用参数化查询：

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

❌ 错误做法——字符串拼接：

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

ORM 会自动对参数进行转义，所以可以放心使用。

5.2 事务管理要点

原生驱动和 SQLAlchemy 默认 autocommit=False，这其实是好事——只有在显式调用 commit() 时数据才会真正写入。任何异常都应及时调用 rollback()，保证数据一致性。本文提供的上下文管理器已经内置了此逻辑，直接拿来用即可。

5.3 索引设计原则

  • 常用查询条件优先建索引：如 WHERE, JOIN, ORDER BY 中的字段。
  • 唯一约束自动建索引：上述模型中的 email 便是典型例子。
  • 避免在小表或极度频繁更新的字段上建索引，可能得不偿失。
  • PostgreSQL 支持 CREATE INDEX CONCURRENTLY，可在不锁表的情况下在线创建索引。
1. **轻量或学习项目**：使用 **原生驱动 + 连接池**，简单直接，依赖少。 2. **中型以上或需要跨库**：直接用 **SQLAlchemy ORM**，降低维护成本。 3. **任何生产环境**：务必做到 **环境变量管理敏感信息** 和 **启用连接池**。

相关教程

总结

Python 连接 MySQL 和 PostgreSQL 的核心逻辑基本相同，差异主要体现在驱动语法以及 PostgreSQL 的独有特性（如 UPSERT、JSONB）。记住连接池、参数化查询、显式事务管理这三大关键点，你就能写出高效、安全、可维护的数据库代码。希望这篇指南能成为你日常开发的速查手册。