怎么使用Python连接MySQL

发布时间：2022-01-25 09:25:31 作者：iii
来源：亿速云阅读：162

# 怎么使用Python连接MySQL

## 前言

在现代数据驱动的应用中，数据库是不可或缺的核心组件。MySQL作为最流行的开源关系型数据库之一，与Python的结合为开发者提供了强大的数据处理能力。本文将详细介绍如何使用Python连接MySQL数据库，涵盖从环境准备到高级操作的完整流程。

---

## 目录
1. [环境准备](#环境准备)
2. [安装必要的库](#安装必要的库)
3. [建立数据库连接](#建立数据库连接)
4. [执行SQL查询](#执行sql查询)
5. [事务管理](#事务管理)
6. [使用ORM框架](#使用orm框架)
7. [连接池技术](#连接池技术)
8. [安全注意事项](#安全注意事项)
9. [常见问题解决](#常见问题解决)
10. [总结](#总结)

---

## 环境准备

### 1.1 安装MySQL服务器
在开始之前，请确保已安装MySQL服务器：
- Windows: 下载MySQL Installer
- macOS: `brew install mysql`
- Linux: `sudo apt-get install mysql-server`

### 1.2 创建测试数据库
```sql
CREATE DATABASE python_mysql;
USE python_mysql;
CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50),
    email VARCHAR(50)
);

安装必要的库

Python通过第三方库实现MySQL连接，主流选择有：

2.1 mysql-connector-python

pip install mysql-connector-python

2.2 PyMySQL

pip install pymysql

2.3 SQLAlchemy（ORM）

pip install sqlalchemy

建立数据库连接

3.1 基础连接示例

import mysql.connector

config = {
    'user': 'root',
    'password': 'yourpassword',
    'host': '127.0.0.1',
    'database': 'python_mysql',
    'raise_on_warnings': True
}

try:
    conn = mysql.connector.connect(**config)
    print("连接成功！")
    conn.close()
except mysql.connector.Error as err:
    print(f"连接失败: {err}")

3.2 使用PyMySQL连接

import pymysql

connection = pymysql.connect(
    host='localhost',
    user='root',
    password='yourpassword',
    database='python_mysql'
)

执行SQL查询

4.1 查询操作

cursor = conn.cursor(dictionary=True)  # 返回字典形式结果
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()

for row in results:
    print(f"ID: {row['id']}, Name: {row['name']}")

4.2 插入数据

insert_query = "INSERT INTO users (name, email) VALUES (%s, %s)"
values = ('Alice', 'alice@example.com')

cursor.execute(insert_query, values)
conn.commit()  # 必须提交事务
print(f"插入行数: {cursor.rowcount}")

4.3 批量插入

data = [
    ('Bob', 'bob@example.com'),
    ('Charlie', 'charlie@example.com')
]
cursor.executemany(insert_query, data)

事务管理

5.1 基本事务

try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")
    conn.commit()
except:
    conn.rollback()

5.2 使用上下文管理器

with conn.cursor() as cursor:
    cursor.execute("BEGIN")
    # 执行多个操作
    conn.commit()

使用ORM框架

6.1 SQLAlchemy基础

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = create_engine('mysql+pymysql://root:password@localhost/python_mysql')

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(50))

# 查询示例
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

users = session.query(User).filter(User.name.like('A%')).all()

连接池技术

7.1 使用DBUtils

from dbutils.pooled_db import PooledDB
import pymysql

pool = PooledDB(
    creator=pymysql,
    maxconnections=10,
    host='localhost',
    user='root',
    password='yourpassword',
    database='python_mysql'
)

conn = pool.connection()
cursor = conn.cursor()
# 执行操作...
conn.close()  # 实际返回到连接池

安全注意事项

8.1 防止SQL注入

# 错误示范
cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# 正确方式
cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))

8.2 连接安全配置

使用SSL连接
限制数据库用户权限
定期更换密码

常见问题解决

9.1 连接超时

config = {
    'connect_timeout': 10,
    'connection_timeout': 5
}

9.2 字符编码问题

charset='utf8mb4',
collation='utf8mb4_unicode_ci'

9.3 错误代码参考

2003: 连接拒绝
1045: 访问被拒绝
2006: 服务器断开连接

总结

本文详细介绍了Python连接MySQL的多种方法，包括： 1. 使用原生驱动进行基础操作 2. 通过ORM框架简化开发 3. 连接池优化性能 4. 安全最佳实践

建议根据项目需求选择合适的方案： - 小型项目：PyMySQL + 直接SQL - 中型项目：SQLAlchemy Core - 大型项目：SQLAlchemy ORM + 连接池

通过合理使用这些技术，可以构建高效可靠的数据库应用。

附录：完整示例代码

# 综合示例：CRUD操作
import mysql.connector
from mysql.connector import Error

def create_connection():
    try:
        conn = mysql.connector.connect(
            host='localhost',
            user='root',
            password='yourpassword',
            database='python_mysql'
        )
        return conn
    except Error as e:
        print(f"Error: {e}")
        return None

def create_user(conn, name, email):
    query = "INSERT INTO users(name, email) VALUES(%s, %s)"
    cursor = conn.cursor()
    try:
        cursor.execute(query, (name, email))
        conn.commit()
        return cursor.lastrowid
    except Error as e:
        print(f"Error: {e}")
        return None

# 其他CRUD函数...

注意：实际使用时请替换数据库凭据并处理异常 “`