3. Python数据库连接：使用sqlite3和pymysql连接数据库，执行原生SQL语句

说实话，很多初学者一上来就扎进ORM的怀抱，觉得写原生SQL太low。但我个人习惯是，先把手伸进泥巴里，摸清楚数据库到底怎么跟你对话。你想想看，ORM再漂亮，底层不还是拼SQL字符串吗？

这一章我们就来干两件事：用Python连上SQLite和MySQL，然后亲手执行几条SQL语句。不绕弯子，直接上代码。

3.1 为什么先学原生SQL？

我在项目中遇到过好几次这样的场景：线上出了个慢查询，DBA丢过来一条几千行的SQL，让我解释业务逻辑。如果我只懂ORM的链式调用，那一刻真的会懵掉。

说白了，原生SQL是数据库的母语。你学会了它，不管换什么ORM、换什么框架，底层逻辑都看得透。而且有些复杂查询——比如多层嵌套的窗口函数、递归CTE——ORM写起来反而更痛苦。

3.2 sqlite3：零配置的本地数据库

SQLite是Python自带的，不需要安装任何东西。它把整个数据库存在一个文件里，特别适合做原型验证、本地测试，或者单机小工具。

适用场景： 开发环境调试、单元测试、嵌入式设备、数据量不大的桌面应用。

3.2.1 连接与创建表

import sqlite3

# 连接数据库（如果文件不存在，会自动创建）
conn = sqlite3.connect('demo.db')

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

# 执行建表语句
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        email TEXT UNIQUE
    )
''')

# 提交事务
conn.commit()

# 关闭连接
conn.close()

嗯，这里要注意：connect() 如果传的是文件名，就会在当前目录下生成一个 .db 文件。如果你传 :memory:，数据库就跑在内存里，程序一结束就没了——我刚开始学的时候经常忘记这点，调试半天数据不见了。

3.2.2 插入与查询

conn = sqlite3.connect('demo.db')
cursor = conn.cursor()

# 插入单条数据
cursor.execute(
    "INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
    ('张三', 28, 'zhangsan@example.com')
)

# 插入多条数据
users_data = [
    ('李四', 32, 'lisi@example.com'),
    ('王五', 25, 'wangwu@example.com')
]
cursor.executemany(
    "INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
    users_data
)

conn.commit()

# 查询所有数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)

conn.close()
小技巧： 使用 ? 作为占位符，而不是用 f-string 拼接字符串。这能防止SQL注入攻击。我曾经见过有人直接写 f"SELECT * FROM users WHERE name='{user_input}'"，结果被注入了删库语句……别学他。

3.3 pymysql：连接MySQL数据库

MySQL是生产环境最常见的数据库之一。pymysql是一个纯Python实现的MySQL客户端，安装简单，用法跟sqlite3非常像。

先安装：

pip install pymysql

3.3.1 连接参数

参数 说明 示例值
host 数据库主机地址 'localhost' 或 '192.168.1.100'
port 端口号 3306
user 用户名 'root'
password 密码 'your_password'
database 数据库名 'test_db'
charset 字符集 'utf8mb4'

3.3.2 连接与操作

import pymysql

# 建立连接
conn = pymysql.connect(
    host='localhost',
    port=3306,
    user='root',
    password='your_password',
    database='test_db',
    charset='utf8mb4'
)

cursor = conn.cursor()

# 创建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS products (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(100) NOT NULL,
        price DECIMAL(10, 2),
        stock INT DEFAULT 0
    )
''')

# 插入数据
cursor.execute(
    "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)",
    ('笔记本电脑', 5999.00, 50)
)

conn.commit()

# 查询
cursor.execute("SELECT * FROM products")
result = cursor.fetchone()  # 取一条
print(result)

cursor.close()
conn.close()
注意： pymysql 的占位符是 %s，不是 sqlite3 的 ?。而且无论字段是什么类型，都用 %s，pymysql 会自动帮你做类型转换。我第一次从 sqlite3 切到 pymysql 时，习惯性写了 ?，结果报错半天才反应过来。

3.4 核心知识体系

下面这张图帮你理清本章的知识脉络：

Python数据库连接核心流程 Python 应用 连接模块：sqlite3 / pymysql SQLite 文件数据库（demo.db） MySQL 服务端数据库（localhost:3306） 执行原生SQL → cursor.execute() / executemany()

3.5 避坑指南与最佳实践

  • 别忘了 commit：sqlite3 和 pymysql 默认都是手动提交事务。你执行了 INSERT 或 UPDATE 后，必须调用 conn.commit()，否则数据不会真正写入。我曾经调试一个数据导入脚本，跑了半小时发现一条都没写进去——就是因为忘了 commit。
  • 用完记得关连接：数据库连接是宝贵资源。建议用 with 语句自动管理上下文，或者至少保证 conn.close() 被执行。
  • 参数化查询是底线：永远不要用字符串拼接构造SQL。这不是风格问题，是安全问题。
  • fetch 方法的选择：fetchone() 取一条，fetchall() 取全部，fetchmany(size) 取指定条数。数据量大时用 fetchmany 分批处理，避免内存爆炸。
我的习惯： 在开发阶段，我会把 conn.row_factory = sqlite3.Row 设置上（sqlite3 支持），这样查询结果可以用列名访问，比如 row['name']，代码可读性会好很多。pymysql 也有类似的 pymysql.cursors.DictCursor。

3.6 完整示例：一个简单的用户管理

import sqlite3

def init_db():
    conn = sqlite3.connect('users.db')
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            age INTEGER,
            city TEXT
        )
    ''')
    conn.commit()
    return conn

def add_user(conn, name, age, city):
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO users (name, age, city) VALUES (?, ?, ?)",
        (name, age, city)
    )
    conn.commit()
    print(f"用户 {name} 添加成功，ID: {cursor.lastrowid}")

def query_users(conn, city=None):
    cursor = conn.cursor()
    if city:
        cursor.execute("SELECT * FROM users WHERE city = ?", (city,))
    else:
        cursor.execute("SELECT * FROM users")
    return cursor.fetchall()

# 使用示例
conn = init_db()
add_user(conn, '赵六', 30, '北京')
add_user(conn, '钱七', 22, '上海')

users = query_users(conn, '北京')
for user in users:
    print(f"{user['name']}, {user['age']}岁, 来自{user['city']}")

conn.close()

你看，代码其实就那么几行。核心就是三步：连接 → 执行 → 关闭。不管换什么数据库，这个骨架都不会变。

下一章我们会在这个基础上，聊聊怎么用 ORM 让代码更优雅。但请记住今天亲手写 SQL 的感觉——这是你理解数据库的根基。

专注资料整理
← 上一章 📖 返回目录 下一章 →