Skip to main content

Python 数据库使用

Python 使用 MySQL 数据库

数据库在现代应用程序中至关重要，而 MySQL 以其高效和易用性受到开发者青睐。Python 通过简单的语法和强大的库，让数据操作变得更轻松。本教程将帮助您掌握 Python 与 MySQL 的结合应用，适用于数据存储、查询和分析。

MySQL 基础

在操作 MySQL 之前，你需要先安装 MySQL 数据，安装 MySQL 不在本教程的范畴之内，请自行解决。

安装好MySQL后你回得到MySQL的地址、端口、用户名、密码。

-- 创建数据库
CREATE DATABASE mydatabase;

-- 使用数据库
USE mydatabase;

-- 创建表
CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

此示例创建了一个名为 mydatabase 的数据库和一个 users 表，包含用户 ID、用户名、邮箱和创建时间。

Python 连接 MySQL

要用 Python 操作 MySQL，推荐使用 mysql-connector-python 库。先安装它：

pip install mysql-connector-python

以下是连接 MySQL 并查询数据的完整代码：

import mysql.connector

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

# 创建游标
cursor = connection.cursor()

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

# 获取结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭游标和连接
cursor.close()
connection.close()

此代码连接到 mydatabase，查询 users 表并打印结果。注意替换 your_username 和 your_password 为您的实际凭据。

事务管理

事务是一组操作，要么全部成功，要么全部失败，确保数据一致性。以下是 Python 中管理事务的示例：

import mysql.connector

connection = mysql.connector.connect(
    host="localhost",
    user="your_username",
    password="your_password",
    database="mydatabase"
)

try:
    # 开始事务
    connection.start_transaction()

    cursor = connection.cursor()

    # 执行插入操作
    cursor.execute("INSERT INTO users (username, email) VALUES ('user1', 'user1@example.com')")
    cursor.execute("INSERT INTO users (username, email) VALUES ('user2', 'user2@example.com')")

    # 提交事务
    connection.commit()

except mysql.connector.Error as error:
    # 回滚事务
    connection.rollback()
    print(f"事务失败: {error}")

finally:
    cursor.close()
    connection.close()

此示例插入两条记录，若成功则提交，否则回滚。

连接池

连接池通过预创建连接提高性能，避免频繁建立和关闭连接。以下是使用连接池的示例：

import mysql.connector
from mysql.connector import pooling

# 配置连接池
dbconfig = {
    "host": "localhost",
    "user": "your_username",
    "password": "your_password",
    "database": "mydatabase"
}
connection_pool = pooling.MySQLConnectionPool(pool_name="mypool", pool_size=5, **dbconfig)

# 获取连接
connection = connection_pool.get_connection()

# 创建游标
cursor = connection.cursor()

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

# 获取结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭游标和连接
cursor.close()
connection.close()

此代码创建了一个大小为 5 的连接池，从中获取连接执行查询。

错误处理

数据库操作可能出错，如连接失败或语法错误。以下是处理错误的示例：

import mysql.connector

try:
    connection = mysql.connector.connect(
        host="localhost",
        user="your_username",
        password="your_password",
        database="mydatabase"
    )

    cursor = connection.cursor()

    # 执行错误查询
    cursor.execute("SELECT * FROM non_existent_table")

except mysql.connector.Error as error:
    print(f"数据库错误: {error}")

finally:
    if 'cursor' in locals():
        cursor.close()
    if 'connection' in locals():
        connection.close()

此代码尝试查询不存在的表，捕获并显示错误信息。