位置：文档库 > Python > Python读取MySQL表数据的方法介绍

Python读取MySQL表数据的方法介绍

RogueCrest 上传于 2022-07-07 20:31

《Python读取MySQL表数据的方法介绍》

在数据驱动的现代应用中，数据库与编程语言的交互能力至关重要。MySQL作为最流行的开源关系型数据库之一，与Python的结合为数据处理提供了高效解决方案。本文将系统介绍Python读取MySQL表数据的多种方法，涵盖从基础连接到高级优化的完整流程，帮助开发者快速掌握核心技能。

一、环境准备与基础连接

在开始操作前，需确保系统已安装MySQL数据库和Python环境。推荐使用Python 3.6+版本，并通过pip安装必要的驱动库：

pip install mysql-connector-python pymysql sqlalchemy

其中mysql-connector-python是MySQL官方驱动，PyMySQL是第三方轻量级驱动，SQLAlchemy则提供ORM高级功能。选择驱动时需考虑项目需求：官方驱动兼容性最佳，PyMySQL安装更简单，SQLAlchemy适合复杂项目。

基础连接示例（使用mysql-connector）：

import mysql.connector

# 建立连接
conn = mysql.connector.connect(
    host="localhost",
    user="your_username",
    password="your_password",
    database="your_database"
)

# 创建游标对象
cursor = conn.cursor()

# 执行SQL查询
cursor.execute("SELECT * FROM employees")

# 获取结果
results = cursor.fetchall()
for row in results:
    print(row)

# 关闭连接
cursor.close()
conn.close()

这段代码演示了连接数据库、执行查询和关闭连接的标准流程。实际开发中，建议将连接参数配置在环境变量或配置文件中，避免硬编码敏感信息。

二、数据查询的多种方式

1. 基础查询方法

fetchone()：逐行获取结果，适合大数据量

cursor.execute("SELECT name FROM customers")
row = cursor.fetchone()
while row is not None:
    print(row[0])
    row = cursor.fetchone()

fetchmany(size)：分批获取结果

cursor.execute("SELECT * FROM products")
batch_size = 100
while True:
    rows = cursor.fetchmany(batch_size)
    if not rows:
        break
    for row in rows:
        process(row)

2. 参数化查询

为防止SQL注入，必须使用参数化查询：

# 正确方式
query = "SELECT * FROM orders WHERE order_date > %s"
cursor.execute(query, ('2023-01-01',))

# 错误方式（存在注入风险）
# cursor.execute(f"SELECT * FROM orders WHERE order_date > '{date}'")

不同驱动的参数占位符不同：mysql-connector使用%s，PyMySQL也使用%s，而psycopg2（PostgreSQL驱动）使用%s但行为略有不同。

三、高级数据处理技术

1. 使用上下文管理器

Python的with语句可自动管理资源：

import mysql.connector

config = {
    'user': 'username',
    'password': 'password',
    'host': 'localhost',
    'database': 'testdb'
}

with mysql.connector.connect(**config) as conn:
    with conn.cursor(dictionary=True) as cursor:
        cursor.execute("SELECT * FROM departments")
        for dept in cursor:
            print(dept['dept_name'])

dictionary=True参数使结果以字典形式返回，可通过列名访问字段。

2. 批量操作优化

处理大量数据时，executemany()可显著提高性能：

data = [
    ('John', 'Doe', '2023-01-15'),
    ('Jane', 'Smith', '2023-02-20'),
    ('Bob', 'Johnson', '2023-03-10')
]

query = "INSERT INTO employees (first_name, last_name, hire_date) VALUES (%s, %s, %s)"
cursor.executemany(query, data)
conn.commit()

3. 事务处理

确保数据一致性：

try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception as e:
    conn.rollback()
    print(f"Transaction failed: {e}")

四、ORM方法：SQLAlchemy核心应用

SQLAlchemy提供了更Pythonic的数据库操作方式：

1. 基础配置

from sqlalchemy import create_engine, MetaData, Table, select

# 创建引擎（echo=True可显示SQL语句）
engine = create_engine('mysql+pymysql://user:pass@localhost/dbname', echo=True)

# 反射获取表结构
metadata = MetaData()
employees = Table('employees', metadata, autoload_with=engine)

2. 查询操作

from sqlalchemy import select

# 简单查询
stmt = select(employees).where(employees.c.salary > 5000)
with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(row)

# 多表连接
from sqlalchemy import join
departments = Table('departments', metadata, autoload_with=engine)
stmt = select(employees, departments).select_from(
    join(employees, departments, employees.c.dept_id == departments.c.id)
)

3. 会话管理

更完整的ORM示例：

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Employee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    salary = Column(Float)

# 创建会话类
Session = sessionmaker(bind=engine)
session = Session()

# 查询
emp = session.query(Employee).filter_by(salary=6000).first()
print(emp.name)

# 添加新记录
new_emp = Employee(name='Alice', salary=5500)
session.add(new_emp)
session.commit()

五、性能优化策略

1. 连接池配置

使用SQLAlchemy的连接池：

from sqlalchemy import create_engine

engine = create_engine(
    'mysql+pymysql://user:pass@localhost/dbname',
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=3600
)

2. 查询优化技巧

只选择需要的列，避免SELECT *
为常用查询条件创建索引
大数据量时使用分页查询

# 分页查询示例
page_size = 50
page = 1
offset = (page - 1) * page_size

query = "SELECT * FROM large_table LIMIT %s OFFSET %s"
cursor.execute(query, (page_size, offset))

3. 异步处理方案

对于高并发场景，可考虑异步驱动：

# 使用aiomysql示例
import asyncio
import aiomysql

async def fetch_data():
    conn = await aiomysql.connect(
        host='localhost',
        port=3306,
        user='user',
        password='pass',
        db='dbname'
    )
    async with conn.cursor() as cursor:
        await cursor.execute("SELECT * FROM products")
        result = await cursor.fetchall()
        print(result)
    conn.close()

asyncio.run(fetch_data())

六、常见问题与解决方案

1. 中文乱码问题

解决方案：在连接时指定字符集

conn = mysql.connector.connect(
    ...
    charset='utf8mb4',
    collation='utf8mb4_unicode_ci'
)

2. 连接超时处理

conn = mysql.connector.connect(
    ...
    connect_timeout=10,
    connection_retry_delay=5
)

3. 驱动兼容性问题

不同驱动特性对比：

特性	mysql-connector	PyMySQL	SQLAlchemy
官方支持	是	否	否
Python 2支持	否	是	是
异步支持	否	通过第三方	通过扩展

七、实际应用案例

1. 数据清洗流程

import pandas as pd
import mysql.connector

# 从MySQL读取数据到DataFrame
def mysql_to_dataframe(query):
    conn = mysql.connector.connect(...)
    df = pd.read_sql(query, conn)
    conn.close()
    return df

# 数据清洗示例
df = mysql_to_dataframe("SELECT * FROM raw_data")
df['cleaned_value'] = df['raw_value'].str.strip().str.upper()

# 将处理后的数据写回MySQL
with mysql.connector.connect(...) as conn:
    df.to_sql('cleaned_data', conn, if_exists='replace', index=False)

2. 实时数据监控

import time
import mysql.connector
from datetime import datetime

def monitor_table(interval=60):
    last_id = 0
    while True:
        conn = mysql.connector.connect(...)
        cursor = conn.cursor()
        
        # 只获取新记录
        cursor.execute("SELECT * FROM sensor_data WHERE id > %s ORDER BY id LIMIT 100", (last_id,))
        new_records = cursor.fetchall()
        
        if new_records:
            last_id = new_records[-1][0]
            process_records(new_records)
        
        cursor.close()
        conn.close()
        time.sleep(interval)

def process_records(records):
    for record in records:
        print(f"{datetime.now()}: New data {record}")

八、最佳实践总结

1. 安全实践

始终使用参数化查询
最小权限原则配置数据库用户
敏感信息使用环境变量或密钥管理服务

2. 性能建议

重用连接而非频繁创建/关闭
大数据量操作使用批量方法
定期分析慢查询并优化

3. 代码组织

将数据库操作封装在单独模块中
使用配置文件管理连接参数
实现重试机制处理临时故障

关键词：Python、MySQL、数据库连接、SQL查询、参数化查询、SQLAlchemy、ORM、性能优化、连接池、异步处理

简介：本文全面介绍了Python读取MySQL数据库的多种方法，从基础连接建立到高级ORM应用，涵盖查询优化、事务处理、性能调优等核心主题。通过代码示例和实际案例，帮助开发者掌握安全、高效的数据库操作技巧，适用于从简单查询到复杂数据处理的各类场景。

立即下载

Python相关