Python数据库实战：SQLite3深度解析-开发者社区

Python数据库实战：SQLite3深度解析

引言

在Python开发中，SQLite是构建轻量级数据库应用的核心技术。作为一名从Rust转向Python的后端开发者，我深刻体会到sqlite3在嵌入式数据库方面的优势。sqlite3是Python标准库中内置的SQLite数据库接口，提供了简洁的API和完整的功能。

sqlite3核心概念

什么是sqlite3

sqlite3是Python标准库中内置的SQLite数据库接口，具有以下特点：

内置模块：无需额外安装
轻量级：嵌入式数据库，无需独立服务
事务支持：支持ACID事务
完整SQL支持：支持标准SQL语法
跨平台：支持多种操作系统

架构设计

┌─────────────────────────────────────────────────────────────┐ │ sqlite3 架构 │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 应用程序 │───▶│ sqlite3 │───▶│ SQLite文件 │ │ │ │ (Application)│ │ 接口层 │ │ (Database) │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ SQL解析引擎 │ │ │ └──────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘

环境搭建与基础配置

基本连接

import sqlite3 conn = sqlite3.connect('example.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT NOT NULL UNIQUE ) ''') conn.commit() conn.close()

插入数据

import sqlite3 conn = sqlite3.connect('example.db') cursor = conn.cursor() user = ('张三', 'zhangsan@example.com') cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', user) conn.commit() conn.close()

高级特性实战

查询数据

import sqlite3 conn = sqlite3.connect('example.db') cursor = conn.cursor() cursor.execute('SELECT * FROM users') users = cursor.fetchall() for user in users: print(user) cursor.execute('SELECT * FROM users WHERE id = ?', (1,)) user = cursor.fetchone() print(user) conn.close()

更新数据

import sqlite3 conn = sqlite3.connect('example.db') cursor = conn.cursor() cursor.execute('UPDATE users SET name = ? WHERE id = ?', ('李四', 1)) conn.commit() conn.close()

删除数据

import sqlite3 conn = sqlite3.connect('example.db') cursor = conn.cursor() cursor.execute('DELETE FROM users WHERE id = ?', (1,)) conn.commit() conn.close()

实际业务场景

场景一：用户管理

import sqlite3 class UserManager: def __init__(self, db_path): self.db_path = db_path def _get_connection(self): return sqlite3.connect(self.db_path) def create_user(self, name, email): conn = self._get_connection() cursor = conn.cursor() try: cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', (name, email)) conn.commit() return cursor.lastrowid finally: conn.close() def get_user(self, user_id): conn = self._get_connection() cursor = conn.cursor() try: cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,)) return cursor.fetchone() finally: conn.close() def get_all_users(self): conn = self._get_connection() cursor = conn.cursor() try: cursor.execute('SELECT * FROM users') return cursor.fetchall() finally: conn.close()

场景二：事务处理

import sqlite3 def transfer_funds(from_id, to_id, amount): conn = sqlite3.connect('bank.db') cursor = conn.cursor() try: cursor.execute('BEGIN TRANSACTION') cursor.execute('SELECT balance FROM accounts WHERE id = ?', (from_id,)) from_balance = cursor.fetchone()[0] if from_balance < amount: raise ValueError('Insufficient funds') cursor.execute('UPDATE accounts SET balance = balance - ? WHERE id = ?', (amount, from_id)) cursor.execute('UPDATE accounts SET balance = balance + ? WHERE id = ?', (amount, to_id)) conn.commit() return True except Exception as e: conn.rollback() raise e finally: conn.close()

性能优化

使用索引

import sqlite3 conn = sqlite3.connect('example.db') cursor = conn.cursor() cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)') conn.commit() conn.close()

批量操作

import sqlite3 conn = sqlite3.connect('example.db') cursor = conn.cursor() users = [ ('张三', 'zhangsan@example.com'), ('李四', 'lisi@example.com'), ('王五', 'wangwu@example.com') ] cursor.executemany('INSERT INTO users (name, email) VALUES (?, ?)', users) conn.commit() conn.close()