怎么使用Python連接MySQL

# 怎么使用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函數...

注意：實際使用時請替換數據庫憑據并處理異常 “`