当前位置：首页 > news >正文

SQLite3 操作指南：SQL 语句与 ORM 方法对比解析

news 2025/9/4 6:36:57

作为开发者，数据存储是项目开发中绕不开的环节。而 SQLite3 凭借其轻量、无需服务端、零配置的特性，成为中小型项目、本地工具及嵌入式应用的首选数据库。在 Python 生态中操作 SQLite3，最常见的有两种方式：直接编写 SQL 语句和使用 ORM（对象关系映射）框架。前者灵活直观，后者则能让我们用面向对象的思维操作数据库，大幅提升代码可读性与可维护性。今天这篇文章，就带大家从实践出发，深入对比这两种操作方式的优劣与适用场景。

一、基础准备：SQLite3 与 Python 的 “第一次握手”

在开始前，我们需要明确：Python 标准库已内置sqlite3模块，无需额外安装即可使用；而 ORM 工具则需要自行安装，本文以最流行的轻量级 ORM——SQLAlchemy为例（需通过pip install sqlalchemy安装）。

首先，我们先定义一个核心业务场景：管理一个 “用户信息表”（users），包含字段：id（主键，自增）、username（用户名，唯一）、email（邮箱）、create_time（创建时间）。后续所有操作，都围绕这个表的 “增删改查”（CRUD）展开。

二、原生 SQL 操作：直接与数据库 “对话”

原生 SQL 的核心思路是：通过sqlite3模块建立数据库连接，创建游标，然后执行手写的 SQL 语句，最后处理结果。这种方式的优势是 “所见即所得”，完全掌控 SQL 逻辑，适合复杂查询场景。

1. 完整实现代码

import sqlite3
import datetime
from sqlite3 import Errordef create_connection(db_file):"""创建数据库连接"""conn = Nonetry:# 连接数据库（不存在则自动创建）conn = sqlite3.connect(db_file)print(f"SQLite3 连接成功（版本：{sqlite3.version}）")# 创建游标cursor = conn.cursor()# 1. 创建users表create_table_sql = """CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT,username TEXT NOT NULL UNIQUE,email TEXT NOT NULL,create_time DATETIME NOT NULL);"""cursor.execute(create_table_sql)conn.commit()print("表创建成功（或已存在）")# 2. 插入一条用户数据（使用参数化查询，避免SQL注入）insert_sql = "INSERT OR IGNORE INTO users (username, email, create_time) VALUES (?, ?, ?);"user_data = ("zhangsan", "zhangsan@example.com", datetime.datetime.now())cursor.execute(insert_sql, user_data)conn.commit()print(f"插入数据成功（影响行数：{cursor.rowcount}）")# 3. 查询所有用户select_sql = "SELECT * FROM users;"cursor.execute(select_sql)users = cursor.fetchall()  # 获取所有结果print("\n所有用户数据：")for user in users:print(f"ID: {user[0]}, 用户名: {user[1]}, 邮箱: {user[2]}, 创建时间: {user[3]}")# 4. 更新用户邮箱update_sql = "UPDATE users SET email = ? WHERE username = ?;"update_data = ("zhangsan_new@example.com", "zhangsan")cursor.execute(update_sql, update_data)conn.commit()print(f"\n更新数据成功（影响行数：{cursor.rowcount}）")# 5. 删除用户（演示用，实际需谨慎）delete_sql = "DELETE FROM users WHERE username = ?;"delete_data = ("lisi",)  # 若表中无lisi，影响行数为0cursor.execute(delete_sql, delete_data)conn.commit()print(f"删除数据成功（影响行数：{cursor.rowcount}）")except Error as e:print(f"操作出错：{e}")finally:# 关闭连接if conn:conn.close()print("\n数据库连接已关闭")# 执行函数（连接到当前目录下的test.db）
create_connection("test.db")

2. 原生 SQL 的优劣势分析

优势：

（1）灵活性极高：支持所有 SQLite3 语法，复杂查询（如多表联查、子查询、聚合函数）可直接编写；

（2）性能损耗低：无需 ORM 的 “翻译” 过程，执行效率更高；

（3）学习成本低：只要掌握 SQL 基础，就能快速上手，无需额外学习 ORM 框架。

劣势：

（1）代码冗余：重复编写连接、游标、提交等模板代码；

（2）维护成本高：表结构变更时，需手动修改所有相关 SQL 语句；

（3）风险较高：若直接拼接字符串，易引发 SQL 注入（需手动使用参数化查询规避）；

（4）可读性差：大量 SQL 语句与 Python 代码混合，后期难以梳理业务逻辑。

三、ORM 操作：用面向对象思维操作数据库

ORM（Object-Relational Mapping）的核心是 “映射”：将数据库表映射为 Python 类，表字段映射为类属性，CRUD 操作映射为类的方法。以SQLAlchemy为例，我们无需手写 SQL，只需定义 “模型类”，即可用 Python 语法操作数据库。

1. 完整实现代码（SQLAlchemy ORM）

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import datetime# 1. 初始化基础配置
# 数据库连接地址（SQLite3格式：sqlite:///数据库路径）
SQLALCHEMY_DATABASE_URL = "sqlite:///test_orm.db"# 创建引擎（SQLite3需添加check_same_thread=False，避免线程问题）
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)# 创建会话工厂（每次操作数据库需创建一个会话）
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 创建模型基类（所有模型类需继承此类）
Base = declarative_base()# 2. 定义User模型类（对应users表）
class User(Base):__tablename__ = "users"  # 数据库表名# 定义字段（与表结构一一对应）id = Column(Integer, primary_key=True, autoincrement=True)username = Column(String, unique=True, nullable=False)  # 唯一、非空email = Column(String, nullable=False)create_time = Column(DateTime, nullable=False)# 可选：定义__repr__方法，方便打印对象时查看信息def __repr__(self):return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"# 3. 创建数据库表（若不存在则创建）
Base.metadata.create_all(bind=engine)
print("表创建成功（或已存在）")# 4. 核心CRUD操作（通过会话实现）
def db_operations():# 创建一个会话（相当于原生SQL的游标）db = SessionLocal()try:# 2. 插入用户（创建User对象，添加到会话，再提交）new_user = User(username="zhangsan",email="zhangsan@example.com",create_time=datetime.datetime.now())db.add(new_user)  # 添加到会话db.commit()       # 提交到数据库db.refresh(new_user)  # 刷新对象，获取自增的idprint(f"插入用户成功：{new_user}")# 3. 查询用户（支持链式调用，无需手写SQL）# 方式1：查询单个用户（按条件）user = db.query(User).filter(User.username == "zhangsan").first()print(f"\n查询单个用户：{user}")# 方式2：查询所有用户all_users = db.query(User).all()print("\n所有用户：")for u in all_users:print(u)# 4. 更新用户（修改对象属性，提交会话）if user:user.email = "zhangsan_new@example.com"db.commit()db.refresh(user)print(f"\n更新用户成功：{user}")# 5. 删除用户（删除对象，提交会话）user_to_delete = db.query(User).filter(User.username == "lisi").first()if user_to_delete:db.delete(user_to_delete)db.commit()print(f"\n删除用户成功：{user_to_delete}")else:print("\n无匹配用户，删除操作跳过")finally:# 关闭会话db.close()print("\n数据库会话已关闭")# 执行ORM操作
db_operations()