Python数据库操作实战:从SQLite到MySQL的CRUD与事务管理
2026/7/31 8:39:35 网站建设 项目流程

1. 项目概述:从脚本到数据管家

如果你已经跟着这个系列走过了前六篇,从打印“Hello World”到能写个简单的爬虫或者处理个Excel文件,那你可能会发现一个问题:我们处理的数据好像都是“一次性”的。程序一关,数据就没了;或者数据稍微多一点,用列表、字典存着就开始卡顿。这时候,数据库就该登场了。它就像一个专门为数据设计的、功能强大的文件柜,能帮你把数据规规矩矩地存起来,想查就查,想改就改,还能保证多人同时操作不乱套。

这篇我们就来聊聊怎么用Python当这个“文件柜”的管理员。核心就两件事:第一,学会用Python连接并操作几种主流的数据库(比如MySQL、SQLite);第二,掌握指挥这个文件柜的“标准口令”——SQL语言。别被“SQL”吓到,你可以把它理解成一套和数据库沟通的固定句式,比如“把姓张的用户都找出来”、“给所有商品价格打八折”。Python的数据库操作库,就是帮你把Python代码翻译成这些SQL句子的翻译官。

我见过不少新手,要么一头扎进复杂的SQL语法里出不来,要么光会用Python的ORM(对象关系映射)工具点点鼠标,底层一问三不知。咱们不走极端,这篇的目标是让你既能用Python流畅地完成“增删改查”这些日常操作,又能明白背后那条SQL命令到底在干什么,做到心里有数,出了问题也知道去哪儿排查。

2. 核心思路:连接、交互与翻译

操作数据库,无论用什么编程语言,其核心逻辑都是一个三层模型:连接层、交互层和数据层。Python在这个模型中扮演的是“交互层”的驱动者和“翻译官”的角色。

2.1 核心模型解析

首先,你需要一个数据库驱动。这就像你要和一位外国朋友交流,你需要一个翻译,或者自己学会他的语言。对于MySQL,这个“翻译”通常是PyMySQLmysql-connector-python;对于PostgreSQL,是psycopg2;而对于轻量级的SQLite,Python标准库sqlite3自带了这个“翻译”功能。驱动负责底层网络通信、数据封包和解包,建立一条从你的Python程序到数据库服务器的可靠通道。

建立连接后,就进入了交互环节。这个环节的核心是Cursor(游标)对象。你可以把游标想象成你伸进数据库“文件柜”里的一只手。你的所有操作——取数据、放数据、修改数据——都需要通过这只“手”来完成。你通过游标执行SQL命令,也通过游标获取返回的结果。理解游标是理解Python数据库操作的关键。

最后是数据层,即SQL语言本身。SQL是一种声明式语言,你只需要告诉数据库你想要什么(“找出所有销售额大于1000的订单”),而不需要指挥它一步步怎么去翻找。Python数据库库的核心任务,就是把你的操作意图(通过函数调用表达)或者你直接编写的SQL字符串,翻译成数据库能听懂的SQL语句,发送出去,再把返回的数据翻译成Python的数据结构(如列表、元组、字典)给你。

2.2 方案选型:DB-API与ORM

Python社区定义了一个操作数据库的标准,叫做Python DB-API 2.0。像sqlite3PyMySQL这些驱动,都遵循这个标准。这意味着你学会了其中一种的基本用法,切换到另一种数据库(在基础操作上)会非常容易,因为它们提供的接口(connect(),cursor(),execute(),fetchall()等)几乎一模一样。我们本篇主要围绕这个标准API展开,这是根基。

而在实际项目中,你可能会遇到ORM,比如SQLAlchemy、Django ORM。ORM的意思是“对象关系映射”,它允许你像操作Python类一样操作数据库表。比如,你定义一个User类,ORM会自动帮你创建对应的用户表,你执行user.save(),它就帮你生成INSERT语句。ORM的优势是开发效率高,代码更“Pythonic”,能避免手写SQL字符串带来的安全风险(如SQL注入)。但它的劣势是,复杂的查询可能不如直接写SQL高效和直观,且隐藏了底层细节,对初学者理解数据库原理不利。

我的建议是:入门阶段,一定要先熟练掌握标准DB-API和原生SQL。这能帮你建立对数据库操作最本质的理解。等你能熟练手写各种JOIN查询、子查询后,再去学习ORM,你会明白ORM在背后帮你做了什么,也能在ORM解决不了性能问题时,有能力直接编写原生SQL进行优化。跳过基础直接上ORM,就像没学会走路就去学跑步,容易摔跤。

3. 环境准备与核心库选择

工欲善其事,必先利其器。我们先来把“翻译官”和“文件柜”准备好。

3.1 数据库选择与安装

对于初学者,我强烈推荐从SQLite开始。理由有三:第一,它无需安装任何服务器软件,数据库就是一个单独的.db文件,随项目携带,极其轻便;第二,Python内置了sqlite3模块,无需额外安装驱动;第三,它支持标准的SQL语法,学会后可以无缝迁移到MySQL等大型数据库。本篇的示例将主要使用SQLite,以确保所有读者都能零成本复现。

当然,我们也会涉及MySQL,因为它是生产环境中最常见的关系型数据库之一。如果你打算跟进MySQL部分,需要先安装MySQL服务器。可以去MySQL官网下载社区版安装包,或者使用更简单的集成工具如XAMPP、MAMP(包含MySQL)。安装完成后,记得启动MySQL服务。

3.2 Python库安装

对于SQLite,无需安装。对于MySQL,我们需要安装Python驱动。这里我推荐PyMySQL,因为它纯Python实现,安装简单,兼容性好。

打开你的终端或命令提示符,使用pip安装:

pip install PyMySQL

如果你想用官方MySQL Connector,可以安装mysql-connector-python,但注意其用法与标准DB-API略有差异。为了遵循通用标准,我们以PyMySQL为例。

3.3 基础连接代码框架

无论操作哪种数据库,连接部分的代码结构都高度相似。下面给出一个通用的、包含异常处理和安全关闭资源的模板,这个模板非常重要,请务必理解每一行的作用。

import sqlite3 # 如果是MySQL,则: import pymysql def create_connection(): """创建数据库连接""" conn = None try: # SQLite连接方式 conn = sqlite3.connect('my_database.db') # 数据库文件,不存在则会自动创建 # MySQL连接方式 (取消注释并修改相应参数) # conn = pymysql.connect( # host='localhost', # 数据库服务器地址 # user='your_username', # 用户名 # password='your_password', # 密码 # database='your_database', # 数据库名 # charset='utf8mb4' # 字符编码,推荐utf8mb4以支持完整Unicode(如表情符号) # ) print("数据库连接成功!") return conn except Exception as e: # 这里捕获的是连接阶段的异常,比如网络不通、密码错误、数据库不存在等 print(f"连接数据库时发生错误: {e}") return None # 使用连接 if __name__ == '__main__': connection = create_connection() if connection is not None: # 后续所有数据库操作都应在这个if语句块内,或确保连接被正确关闭 # ... 执行查询 ... connection.close() # 非常重要!操作完毕后必须关闭连接 print("连接已关闭。") else: print("无法建立数据库连接,程序退出。")

关键提示try...except块和conn.close()必须的。网络和IO操作随时可能出错,良好的异常处理能让你的程序更健壮。而忘记关闭连接是常见错误,会导致数据库连接资源泄露,在Web服务器等高并发场景下,很快会耗光所有可用连接,导致服务不可用。更优雅的做法是使用with语句上下文管理器,但初学阶段先明确写出close()有助于建立资源管理意识。

4. 核心操作一:执行SQL与创建表

连接建立后,第一件事往往是创建存储数据的“表格”。在关系型数据库中,数据存储在表(Table)中,表由行(记录)和列(字段)组成。定义表结构,就是定义每个字段的名字和数据类型。

4.1 创建游标与执行DDL

DDL(Data Definition Language)是用于定义和修改数据库结构的语言,如CREATE TABLE,ALTER TABLE,DROP TABLE

def create_table(conn): """创建一个用户表""" # 创建游标对象,所有SQL命令都通过游标执行 cursor = conn.cursor() # 定义SQL语句。SQLite的数据类型包括:INTEGER, TEXT, REAL, BLOB等。 # MySQL中常用:INT, VARCHAR(255), TEXT, DATETIME, DECIMAL等。 create_table_sql = """ CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- SQLite的自增语法 -- id INT PRIMARY KEY AUTO_INCREMENT, -- MySQL的自增语法 username TEXT NOT NULL UNIQUE, email TEXT NOT NULL, age INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """ try: cursor.execute(create_table_sql) # 执行SQL语句 conn.commit() # 提交事务,使创建操作生效。对于DDL,某些数据库会自动提交,但显式提交是好习惯。 print("表 'users' 创建成功或已存在。") except Exception as e: print(f"创建表时发生错误: {e}") conn.rollback() # 如果发生错误,回滚事务。对于DDL,在某些数据库上可能无效,但保留此语句是标准做法。 finally: cursor.close() # 关闭游标。游标也是一种资源,使用后应关闭。 # 在主函数中调用 if __name__ == '__main__': conn = create_connection() if conn: create_table(conn) conn.close()

4.2 代码逐行解析与避坑指南

  • cursor = conn.cursor(): 从连接对象获取一个游标。你可以创建多个游标执行不同任务,但通常一个线程用一个游标就够了。
  • CREATE TABLE IF NOT EXISTS: 这是一个非常实用的语法。如果表已存在,则什么都不做,避免报错。在初始化脚本中常用。
  • 字段定义
    • id INTEGER PRIMARY KEY AUTOINCREMENT: 定义id字段为整数、主键、且自动增长。主键唯一标识一条记录。AUTOINCREMENT是SQLite的关键字,在MySQL中是AUTO_INCREMENT
    • NOT NULL: 约束该字段不能为空。
    • UNIQUE: 约束该字段值在整个表中必须唯一。
    • DEFAULT CURRENT_TIMESTAMP: 默认值为当前时间戳。插入记录时如果不指定该字段,数据库会自动填入当前时间。
  • cursor.execute(): 游标的execute方法用于执行一条SQL语句。这里执行的是一条不返回数据的DDL语句。
  • conn.commit():这是关键点!在数据库中,写操作(INSERT, UPDATE, DELETE, DDL)通常在一个“事务”中。commit()表示确认并提交这个事务,使更改永久化。如果不提交,关闭连接后你的更改可能会丢失。
  • conn.rollback(): 如果try块中的任何代码出错(比如SQL语法错误、违反唯一约束),则执行回滚,撤销当前事务中的所有未提交操作,保持数据一致性。
  • cursor.close(): 在finally块中关闭游标,确保无论是否发生异常,游标资源都会被释放。

实操心得:在开发测试阶段,你可能会反复执行创建表的脚本。使用IF NOT EXISTS可以避免“表已存在”的错误。但在生产环境部署时,更常见的做法是使用数据库迁移工具(如Alembic配合SQLAlchemy,或Django的migrate命令)来管理表结构的变更,这能记录每次变更的历史,并方便地在不同环境间同步。

5. 核心操作二:增删改查(CRUD)实战

CRUD是数据库操作的基石:Create(创建)、Read(读取)、Update(更新)、Delete(删除)。我们结合SQL语句和Python代码来逐一实现。

5.1 插入数据(Create)

users表插入新记录。这里会引入一个极其重要的安全概念:参数化查询

def insert_user(conn, username, email, age=None): """向users表插入一条新用户记录""" cursor = conn.cursor() # 方式一:直接拼接SQL字符串(**危险!切勿在生产环境使用!**) # bad_sql = f"INSERT INTO users (username, email, age) VALUES ('{username}', '{email}', {age})" # 如果username是 `admin' -- `,那么SQL就变成了 `INSERT ... VALUES ('admin' -- ', ...)`,`--`之后的内容被注释掉,可能导致非预期行为或SQL注入攻击。 # 方式二:参数化查询(**安全,推荐!**) sql = "INSERT INTO users (username, email, age) VALUES (%s, %s, %s)" # PyMySQL使用%s作为占位符 # 对于SQLite,占位符是问号(?): "INSERT ... VALUES (?, ?, ?)" # 准备要插入的数据元组 data = (username, email, age) try: cursor.execute(sql, data) # 将数据和SQL分开传入,驱动会安全地处理参数 conn.commit() # 插入数据必须提交事务 print(f"用户 '{username}' 插入成功,ID为: {cursor.lastrowid}") return cursor.lastrowid # 返回刚插入记录的自增ID except Exception as e: # 常见的异常:唯一约束冲突(username重复)、非空约束违反等 print(f"插入用户失败: {e}") conn.rollback() return None finally: cursor.close() # 插入多条数据 def insert_many_users(conn, user_list): """批量插入用户数据,效率远高于循环执行单条INSERT""" cursor = conn.cursor() sql = "INSERT INTO users (username, email, age) VALUES (%s, %s, %s)" try: # executemany 用于批量执行同一条SQL,数据是一个包含多个元组的列表 cursor.executemany(sql, user_list) # user_list 示例: [('张三','zhangsan@xx.com',25), ('李四','lisi@xx.com',30)] conn.commit() print(f"批量插入了 {cursor.rowcount} 条记录。") except Exception as e: print(f"批量插入失败: {e}") conn.rollback() finally: cursor.close()

为什么参数化查询能防SQL注入?因为数据库驱动在接收到参数后,会对参数进行正确的转义和处理,确保它只被当作数据,而不会被解释为SQL代码的一部分。这是Web安全的基础之一,务必养成习惯。

5.2 查询数据(Read)

查询是最常见的操作,游标提供了几种获取结果的方法。

def query_users(conn, min_age=None): """查询用户,可选年龄过滤""" cursor = conn.cursor() # 基础查询 sql = "SELECT id, username, email, age, created_at FROM users" params = () # 动态添加WHERE条件 if min_age is not None: sql += " WHERE age >= %s" # SQLite用 ? params = (min_age,) sql += " ORDER BY created_at DESC" # 按创建时间降序排列 try: cursor.execute(sql, params) # 获取结果的方式: # 1. fetchall(): 获取所有结果行,返回一个列表,列表的每个元素是一个元组,对应一行记录。 # rows = cursor.fetchall() # for row in rows: # print(row) # 例如:(1, '张三', 'zhangsan@xx.com', 25, '2023-10-27 10:00:00') # 2. fetchone(): 获取下一行。常用于只期望一条结果或结果集很大时逐行处理。 # row = cursor.fetchone() # while row is not None: # print(row) # row = cursor.fetchone() # 3. fetchmany(size): 获取指定数量的行。 # rows = cursor.fetchmany(5) # 获取5行 # 更友好的方式:使用字典游标(非标准,但很多驱动支持) # 对于PyMySQL,创建游标时可以指定 cursorclass=pymysql.cursors.DictCursor # 对于sqlite3,可以设置 conn.row_factory = sqlite3.Row,然后使用字典式访问 # 这里演示标准fetchall rows = cursor.fetchall() print(f"查询到 {len(rows)} 条记录:") for row in rows: # 通过索引访问 print(f" ID:{row[0]}, 用户名:{row[1]}, 邮箱:{row[2]}, 年龄:{row[3]}, 注册时间:{row[4]}") # 如果使用了字典游标,可以这样:print(f" ID:{row['id']}, 用户名:{row['username']}...") return rows except Exception as e: print(f"查询失败: {e}") return [] finally: cursor.close()

5.3 更新与删除数据(Update & Delete)

更新和删除操作影响数据,务必谨慎,通常需要结合WHERE条件精确指定目标。

def update_user_email(conn, user_id, new_email): """更新指定用户的邮箱""" cursor = conn.cursor() sql = "UPDATE users SET email = %s WHERE id = %s" data = (new_email, user_id) try: cursor.execute(sql, data) conn.commit() # rowcount属性返回受影响的行数 if cursor.rowcount > 0: print(f"成功更新了 {cursor.rowcount} 条记录(用户ID: {user_id})。") else: print(f"未找到ID为 {user_id} 的用户,更新操作未影响任何记录。") except Exception as e: print(f"更新用户邮箱失败: {e}") conn.rollback() finally: cursor.close() def delete_user(conn, username): """删除指定用户名的用户""" cursor = conn.cursor() sql = "DELETE FROM users WHERE username = %s" data = (username,) try: cursor.execute(sql, data) conn.commit() if cursor.rowcount > 0: print(f"成功删除了 {cursor.rowcount} 条记录(用户名: {username})。") else: print(f"未找到用户名为 '{username}' 的用户。") except Exception as e: print(f"删除用户失败: {e}") conn.rollback() finally: cursor.close()

重要警告UPDATEDELETE语句永远、永远不要忘记写WHERE子句,除非你确实想更新或删除整张表的所有数据。在生产环境执行此类操作前,最好先写一个SELECT语句用相同的WHERE条件确认一下目标数据,例如SELECT * FROM users WHERE username = 'xxx';

6. 事务处理与连接管理进阶

之前我们提到了commit()rollback(),它们都与“事务”有关。事务是数据库保证数据一致性和完整性的核心机制。

6.1 事务的基本概念

事务具有ACID特性:

  • 原子性(Atomicity):事务内的所有操作要么全部成功,要么全部失败回滚。
  • 一致性(Consistency):事务使数据库从一个一致状态转变到另一个一致状态。
  • 隔离性(Isolation):并发事务之间互不干扰。
  • 持久性(Durability):事务一旦提交,其结果就是永久性的。

在Python DB-API中,默认是自动提交模式吗?这取决于驱动和连接设置。对于sqlite3,默认是手动提交模式,即你需要显式调用conn.commit()。对于PyMySQL,默认是自动提交模式,即每条SQL语句都被视为一个独立的事务并立即提交。为了保持行为一致和显式控制,我建议始终显式管理事务

6.2 使用上下文管理器简化操作

手动管理连接的开关和事务的提交回滚比较繁琐,且容易遗漏。Python的with语句(上下文管理器)可以极大地简化这个过程。

对于SQLite,可以这样用:

import sqlite3 # 使用with语句管理连接 with sqlite3.connect('test.db') as conn: # 在这个代码块内,conn是有效的 cursor = conn.cursor() cursor.execute("INSERT INTO users (username) VALUES ('test')") # 不需要显式调用conn.commit(),with块正常退出时会自动提交。 # 如果发生异常,则会自动回滚。 print("操作完成,已自动提交。") # 退出with块后,连接会自动关闭。

对于PyMySQL,它本身没有实现连接的上下文管理器,但我们可以结合try...except...finally或使用第三方库。更常见的做法是封装一个自己的上下文管理器,或者使用ORM(如SQLAlchemy的Session)来管理。

6.3 连接池简介

在Web应用等高频访问数据库的场景下,频繁地创建和关闭数据库连接开销很大。连接池技术应运而生。连接池在程序启动时创建一定数量的数据库连接放在“池”中,当需要时从池中取用一个空闲连接,用完后归还,而不是真正关闭它。

Python中可以使用DBUtilsSQLAlchemy(它内置了连接池)来实现。例如,使用SQLAlchemy的引擎:

from sqlalchemy import create_engine # 连接字符串格式: 数据库类型+驱动://用户名:密码@主机:端口/数据库名 engine = create_engine('mysql+pymysql://user:pass@localhost/mydb?charset=utf8mb4', pool_size=5, # 连接池大小 pool_recycle=3600) # 连接回收时间(秒) # 从连接池获取连接 with engine.connect() as connection: result = connection.execute("SELECT * FROM users") for row in result: print(row) # 连接自动归还到池中

对于初学者,知道这个概念即可。当你的应用从脚本升级到服务时,连接池是必须考虑的部分。

7. 常见问题、性能优化与排查技巧

在实际操作中,你肯定会遇到各种问题和性能瓶颈。这里记录一些典型场景和解决思路。

7.1 常见错误与排查表

错误现象/提示可能原因排查步骤与解决方案
OperationalError: unable to open database file(SQLite)1. 文件路径不存在或无权访问。
2. 磁盘已满。
1. 检查文件路径是否正确,程序是否有该目录的读写权限。
2. 使用绝对路径。检查磁盘空间。
pymysql.err.OperationalError: (2003, “Can‘t connect to MySQL server”)1. MySQL服务未启动。
2. 主机、端口、防火墙配置错误。
3. 用户权限不足。
1. 在系统服务中启动MySQL。
2. 确认hostport(默认3306)正确,防火墙是否放行。
3. 用命令行工具(如mysql -u root -p)测试连接和权限。
pymysql.err.ProgrammingError: (1064, “You have an error in your SQL syntax”)SQL语句语法错误。1. 将打印出的SQL语句复制到数据库客户端(如MySQL Workbench, DBeaver)中直接执行,看具体报错。
2. 检查引号、括号是否配对,关键字是否拼写正确。
3. 注意不同数据库(SQLite vs MySQL)的语法差异,如自增关键字。
pymysql.err.IntegrityError: (1062, “Duplicate entry ‘xxx’ for key ‘username’”)违反了唯一约束,插入了重复的值。1. 检查业务逻辑,确保唯一字段(如用户名)不重复。
2. 插入前可以先查询是否存在(SELECT ... WHERE username=%s),或使用INSERT IGNORE/ON DUPLICATE KEY UPDATE(MySQL)等语法。
pymysql.err.InternalError: (1366, “Incorrect string value”)字符编码问题,尝试存储了不支持的字符(如某些emoji)。1. 确保数据库、表、连接字符串的字符集设置为utf8mb4(MySQL)。
2. Python连接时指定charset='utf8mb4'
查询速度慢,特别是数据量大时1. 没有使用索引。
2. 查询语句写法不佳(如SELECT *)。
3. 频繁建立连接。
1. 在经常用于WHEREJOINORDER BY的字段上创建索引:CREATE INDEX idx_username ON users(username);
2. 只查询需要的列,避免SELECT *
3. 使用连接池复用连接。

7.2 性能优化要点

  1. 使用索引:这是提升查询速度最有效的手段。主键会自动创建索引。为高频查询条件字段创建索引。但注意,索引会降低插入和更新速度(因为要维护索引),且占用额外空间。
  2. 批量操作:如前所述,executemany()比循环execute()快得多。对于大量数据插入,还可以考虑MySQL的LOAD DATA INFILE命令。
  3. 只取所需数据:避免使用SELECT *,明确列出需要的字段。这能减少网络传输的数据量。
  4. 使用连接池:如前所述,在高并发应用中至关重要。
  5. 合理设计数据库结构:遵循数据库设计范式,避免数据冗余和更新异常。这属于更高级的数据库设计知识。

7.3 调试技巧

  • 打印真实SQL:在调试时,有时需要查看驱动最终发送给数据库的SQL语句。对于参数化查询,驱动不会直接给你拼接好的字符串。你可以通过启用数据库的通用查询日志,或者使用驱动的调试选项(如PyMySQL可以在连接时设置cursorclass=pymysql.cursors.DebugCursor)来查看。
  • 使用专业的数据库客户端:如DBeaver、DataGrip、Navicat等。在这些工具中直接编写和测试SQL语句,确认无误后再移植到Python代码中,能极大提高效率。
  • 异常信息细读:数据库返回的错误信息通常很具体,包含了错误代码和描述。仔细阅读,大部分问题都能定位。

8. 从基础到实践:一个小型项目示例

让我们把上面的知识串联起来,实现一个简单的“用户注册登录查询”命令行程序。这个示例将包含创建表、用户注册(插入)、用户登录(查询验证)、查看所有用户等功能。

import sqlite3 import hashlib import getpass # 用于安全输入密码(本例中我们用邮箱简化,实际应用密码需哈希存储) def init_database(): """初始化数据库和表""" conn = sqlite3.connect('user_system.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, email TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, -- 存储密码的哈希值,切勿存明文! created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() print("数据库初始化完成。") def hash_password(password): """简单的密码哈希函数(实际应用应使用更安全的如bcrypt)""" return hashlib.sha256(password.encode()).hexdigest() def register_user(): """用户注册""" username = input("请输入用户名: ").strip() email = input("请输入邮箱: ").strip() password = getpass.getpass("请输入密码: ") # 输入密码时不回显 password_confirm = getpass.getpass("请再次输入密码: ") if password != password_confirm: print("两次输入的密码不一致!") return password_hash = hash_password(password) conn = sqlite3.connect('user_system.db') cursor = conn.cursor() try: cursor.execute( "INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)", (username, email, password_hash) ) conn.commit() print(f"用户 '{username}' 注册成功!") except sqlite3.IntegrityError as e: # 捕获唯一约束违反错误 if 'username' in str(e): print("错误:用户名已存在!") elif 'email' in str(e): print("错误:邮箱已被注册!") else: print(f"注册失败: {e}") except Exception as e: print(f"注册过程中发生未知错误: {e}") conn.rollback() finally: conn.close() def login_user(): """用户登录""" username = input("请输入用户名: ").strip() password = getpass.getpass("请输入密码: ") password_hash = hash_password(password) conn = sqlite3.connect('user_system.db') # 设置行工厂为Row对象,方便通过列名访问 conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute( "SELECT id, username, password_hash FROM users WHERE username = ?", (username,) ) user = cursor.fetchone() # 只期望一条记录 conn.close() if user is None: print("错误:用户名不存在!") elif user['password_hash'] == password_hash: print(f"登录成功!欢迎回来,{user['username']} (ID: {user['id']})。") else: print("错误:密码不正确!") def list_all_users(): """列出所有用户(管理员功能)""" conn = sqlite3.connect('user_system.db') conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT id, username, email, created_at FROM users ORDER BY id") users = cursor.fetchall() conn.close() if not users: print("系统中暂无用户。") return print("\n=== 所有用户列表 ===") print(f"{'ID':<5} {'用户名':<15} {'邮箱':<25} {'注册时间':<20}") print("-" * 70) for user in users: print(f"{user['id']:<5} {user['username']:<15} {user['email']:<25} {user['created_at']:<20}") print(f"总计: {len(users)} 位用户\n") def main_menu(): """主菜单""" init_database() # 程序启动时初始化数据库 while True: print("\n=== 用户管理系统 ===") print("1. 用户注册") print("2. 用户登录") print("3. 查看所有用户") print("4. 退出系统") choice = input("请选择操作 (1-4): ").strip() if choice == '1': register_user() elif choice == '2': login_user() elif choice == '3': list_all_users() elif choice == '4': print("感谢使用,再见!") break else: print("无效选择,请重新输入。") if __name__ == '__main__': main_menu()

这个示例涵盖了之前讲解的大部分核心知识点:连接、DDL、参数化查询、异常处理、事务控制、结果遍历。同时,它也引入了一些实际开发中的考量:

  • 密码安全:绝对不要在数据库中存储明文密码。示例中使用了SHA-256哈希,但在真实项目中,应使用专门为密码存储设计的、加盐的慢哈希函数,如bcryptArgon2
  • 用户体验:简单的命令行交互。
  • 数据展示:格式化输出查询结果。

你可以运行这个程序,体验完整的CRUD流程。试着注册几个用户,然后登录,再查看列表。这是你将Python与数据库结合,迈向构建真实应用的第一步。从这里出发,你可以为其增加更多功能,比如修改用户信息、删除用户、分页显示用户列表等,每一步都是对所学知识的巩固和深化。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询