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

Python数据库操作实战

news 2026/7/30 1:55:07

Python数据库操作实战

后端转 Rust 的萌新，ID “第一程序员”——名字大，人很菜（暂时）。正在跟所有权和生命周期死磕，日常记录 Rust 学习路上的踩坑经验和"啊哈时刻"，代码片段保证能跑。保持学习，保持输出。欢迎大佬们轻喷，也欢迎同好一起进步。

前言

最近在学习 Python 的过程中，我开始关注数据库操作。作为一个从后端转 Rust 的萌新，我认为了解 Python 的数据库操作是非常有必要的，它可以帮助我们与数据库进行交互，存储和管理数据。

Python 提供了多种库和工具来进行数据库操作，如 sqlite3、MySQLdb、psycopg2、SQLAlchemy 等。今天，我就来分享一下 Python 数据库操作的相关知识和实战经验，希望能帮到和我一样的萌新们。

数据库的基本概念

什么是数据库

数据库是一种用于存储和管理数据的系统，它可以帮助我们有效地组织和检索数据。

常见的数据库类型

  • 关系型数据库：如 MySQL、PostgreSQL、SQLite、Oracle 等
  • 非关系型数据库：如 MongoDB、Redis、Cassandra 等

关系型数据库的基本概念

  • 表（Table）：存储数据的基本结构
  • 行（Row）：表中的一条记录
  • 列（Column）：表中的一个字段
  • 主键（Primary Key）：唯一标识表中每条记录的字段
  • 外键（Foreign Key）：关联两个表的字段
  • SQL（Structured Query Language）：用于操作数据库的语言

常用的数据库操作库

1. sqlite3

sqlite3是 Python 内置的 SQLite 数据库接口，用于操作 SQLite 数据库。

importsqlite3# 连接数据库conn=sqlite3.connect('example.db')# 创建游标c=conn.cursor()# 创建表c.execute('''CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)''')# 插入数据c.execute("INSERT INTO users (name, age) VALUES (?, ?)",('Alice',25))c.execute("INSERT INTO users (name, age) VALUES (?, ?)",('Bob',30))# 提交更改conn.commit()# 查询数据c.execute("SELECT * FROM users")print(c.fetchall())# 更新数据c.execute("UPDATE users SET age = ? WHERE name = ?",(26,'Alice'))conn.commit()# 删除数据c.execute("DELETE FROM users WHERE name = ?",('Bob',))conn.commit()# 关闭连接conn.close()

2. psycopg2

psycopg2是 PostgreSQL 数据库的 Python 接口，用于操作 PostgreSQL 数据库。

importpsycopg2# 连接数据库conn=psycopg2.connect(host="localhost",database="test",user="postgres",password="password")# 创建游标cur=conn.cursor()# 创建表cur.execute('''CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name VARCHAR(100), age INTEGER)''')# 插入数据cur.execute("INSERT INTO users (name, age) VALUES (%s, %s)",('Alice',25))cur.execute("INSERT INTO users (name, age) VALUES (%s, %s)",('Bob',30))# 提交更改conn.commit()# 查询数据cur.execute("SELECT * FROM users")print(cur.fetchall())# 关闭游标和连接cur.close()conn.close()

3. SQLAlchemy

SQLAlchemy是一个功能强大的 ORM（对象关系映射）库，用于操作各种数据库。

fromsqlalchemyimportcreate_engine,Column,Integer,Stringfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmaker# 创建引擎engine=create_engine('sqlite:///example.db')# 创建基类Base=declarative_base()# 定义模型classUser(Base):__tablename__='users'id=Column(Integer,primary_key=True)name=Column(String)age=Column(Integer)# 创建表Base.metadata.create_all(engine)# 创建会话Session=sessionmaker(bind=engine)session=Session()# 插入数据user1=User(name='Alice',age=25)user2=User(name='Bob',age=30)session.add_all([user1,user2])session.commit()# 查询数据users=session.query(User).all()foruserinusers:print(f"ID:{user.id}, Name:{user.name}, Age:{user.age}")# 更新数据user=session.query(User).filter_by(name='Alice').first()user.age=26session.commit()# 删除数据user=session.query(User).filter_by(name='Bob').first()session.delete(user)session.commit()# 关闭会话session.close()

实战案例：用户管理系统

1. 设计数据库结构

-- users表CREATETABLEIFNOTEXISTSusers(idINTEGERPRIMARYKEYAUTOINCREMENT,usernameVARCHAR(50)UNIQUENOTNULL,passwordVARCHAR(100)NOTNULL,emailVARCHAR(100)UNIQUENOTNULL,created_atTIMESTAMPDEFAULTCURRENT_TIMESTAMP);-- posts表CREATETABLEIFNOTEXISTSposts(idINTEGERPRIMARYKEYAUTOINCREMENT,user_idINTEGERNOTNULL,titleVARCHAR(200)NOTNULL,contentTEXTNOTNULL,created_atTIMESTAMPDEFAULTCURRENT_TIMESTAMP,FOREIGNKEY(user_id)REFERENCESusers(id));

2. 实现数据库操作

importsqlite3importhashlibfromdatetimeimportdatetimeclassDatabase:def__init__(self,db_name):self.db_name=db_name self.conn=Noneself.cursor=Noneself.connect()self.create_tables()defconnect(self):self.conn=sqlite3.connect(self.db_name)self.cursor=self.conn.cursor()defcreate_tables(self):# 创建users表self.cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username VARCHAR(50) UNIQUE NOT NULL, password VARCHAR(100) NOT NULL, email VARCHAR(100) UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )''')# 创建posts表self.cursor.execute(''' CREATE TABLE IF NOT EXISTS posts ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, title VARCHAR(200) NOT NULL, content TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users (id) )''')self.conn.commit()defhash_password(self,password):returnhashlib.sha256(password.encode()).hexdigest()defadd_user(self,username,password,email):hashed_password=self.hash_password(password)try:self.cursor.execute("INSERT INTO users (username, password, email) VALUES (?, ?, ?)",(username,hashed_password,email))self.conn.commit()returnTrueexceptsqlite3.IntegrityError:returnFalsedefget_user(self,username):self.cursor.execute("SELECT * FROM users WHERE username = ?",(username,))returnself.cursor.fetchone()defverify_password(self,username,password):user=self.get_user(username)ifuser:hashed_password=self.hash_password(password)returnuser[2]==hashed_passwordreturnFalsedefadd_post(self,user_id,title,content):self.cursor.execute("INSERT INTO posts (user_id, title, content) VALUES (?, ?, ?)",(user_id,title,content))self.conn.commit()returnself.cursor.lastrowiddefget_posts(self,user_id=None):ifuser_id:self.cursor.execute("SELECT * FROM posts WHERE user_id = ? ORDER BY created_at DESC",(user_id,))else:self.cursor.execute("SELECT * FROM posts ORDER BY created_at DESC")returnself.cursor.fetchall()defclose(self):ifself.conn:self.conn.close()# 测试if__name__=="__main__":db=Database('user_management.db')# 添加用户db.add_user('admin','password123','admin@example.com')db.add_user('user1','password456','user1@example.com')# 验证用户print(db.verify_password('admin','password123'))# Trueprint(db.verify_password('admin','wrongpassword'))# False# 添加帖子user=db.get_user('admin')ifuser:db.add_post(user[0],'First Post','Hello World!')db.add_post(user[0],'Second Post','How are you?')# 获取帖子posts=db.get_posts()forpostinposts:print(f"ID:{post[0]}, User ID:{post[1]}, Title:{post[2]}, Content:{post[3]}, Created At:{post[4]}")db.close()

数据库操作的最佳实践

1. 使用参数化查询

# 不好的做法cursor.execute(f"SELECT * FROM users WHERE username = '{username}'")# 好的做法cursor.execute("SELECT * FROM users WHERE username = ?",(username,))

2. 关闭数据库连接

# 使用上下文管理器withsqlite3.connect('example.db')asconn:cursor=conn.cursor()# 执行操作# 连接会自动关闭

3. 使用事务

# 开始事务try:# 执行多个操作cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)",('Alice',25))cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)",('Bob',30))# 提交事务conn.commit()except:# 回滚事务conn.rollback()raise

4. 索引优化

-- 创建索引CREATEINDEXidx_users_usernameONusers(username);CREATEINDEXidx_posts_user_idONposts(user_id);

5. 错误处理

try:# 执行数据库操作exceptsqlite3.Errorase:print(f"Database error:{e}")exceptExceptionase:print(f"Error:{e}")finally:# 关闭连接ifconn:conn.close()

常见问题与解决方案

1. 数据库连接失败

问题：无法连接到数据库，出现连接错误。

解决方案：

  • 检查数据库服务器是否运行
  • 检查连接参数是否正确
  • 检查网络连接是否正常
  • 检查数据库权限是否正确

2. 数据插入失败

问题：数据插入失败，出现完整性错误。

解决方案：

  • 检查数据是否符合表结构要求
  • 检查唯一约束是否被违反
  • 检查外键约束是否被违反
  • 使用参数化查询，避免 SQL 注入

3. 查询性能问题

问题：查询速度慢，影响应用性能。

解决方案：

  • 创建适当的索引
  • 优化查询语句
  • 减少查询返回的数据量
  • 使用分页查询
  • 考虑使用缓存

4. 事务处理问题

问题：事务处理不当，导致数据不一致。

解决方案：

  • 使用 try-except-finally 结构处理事务
  • 确保在异常情况下回滚事务
  • 避免长时间持有事务
  • 考虑使用分布式事务（对于复杂系统）

5. 安全问题

问题：数据库操作存在安全风险，如 SQL 注入。

解决方案：

  • 使用参数化查询
  • 对用户输入进行验证和过滤
  • 限制数据库用户的权限
  • 加密敏感数据
  • 定期备份数据库

总结

Python 数据库操作是后端开发的重要组成部分，它可以帮助我们存储和管理数据。通过本文的学习，我们了解了数据库的基本概念、常用的数据库操作库、实战案例、最佳实践和常见问题与解决方案。

作为一个从后端转 Rust 的萌新，我认为学习 Python 的数据库操作是非常有价值的。它不仅可以帮助我们与数据库进行交互，还可以让我们更好地理解数据存储和管理的原理。

在进行数据库操作时，我们应该使用参数化查询、关闭数据库连接、使用事务、优化索引和处理错误。同时，我们还应该注意解决数据库连接失败、数据插入失败、查询性能问题、事务处理问题和安全问题等常见问题。

保持学习，保持输出！今天的 Python 数据库操作实战文章就到这里，希望对大家有所帮助。欢迎在评论区分享你的经验和问题，我们一起进步！

http://www.cnnetsun.cn/news/1824294.html

相关文章：

  • Isaac Sim 8 灯光参数全解析：从零到一的实战调光指南
  • 三步搞定QQ空间历史说说完整备份：GetQzonehistory终极指南
  • 若依与BladeX框架下用户组织架构同步的实践指南
  • 用Chord视频分析工具做影视剪辑：快速定位特定场景与人物出场时间
  • QT桌面应用集成Phi-4-mini-reasoning：开发智能配置向导与帮助系统
  • 如何永久保存QQ空间青春记忆？GetQzonehistory开源工具完整备份指南
  • 怎样高效使用PCB分析工具：硬件工程师的实战指南
  • 数字文旅必备工具：Asian Beauty Z-Image Turbo生成古风虚拟导游全流程
  • 鸿蒙Flutter三方库适配：Flutter Markdown适配实战-鸿蒙平台的Markdown渲染解决方案
  • 博导建议：研究生至少要有一篇 “保底” 论文
  • Qwen3-ASR-1.7B在在线教育中的应用：实时课堂语音转文字
  • Umi-OCR终极指南：如何免费快速完成截图、批量图片和PDF的文字识别
  • 百度网盘秒传脚本终极指南：3分钟学会永久分享文件
  • 2026年东莞墙面防水重做，这些要点你不得不知！
  • GoB插件：5分钟实现Blender与ZBrush无缝桥接的完整指南
  • 【拒绝付费降重】国产大模型立大功！DeepSeek+豆包两步褪去“AI味”，论文AI率80%降至10%通关攻略
  • SVM 面试题总结
  • Betaflight飞控性能优化终极指南：从基础配置到高级调参实战
  • GEM5新手避坑指南：2023最新Docker编译与X86模拟实战（附常见报错解决）
  • AI算法岗和开发岗有什么区别？哪种前景更好？
  • 氢动力飞行器：低空经济的绿色新引擎，技术、应用与市场全解析
  • 别急着回滚！Dify 1.5.0的Markdown文件下载失效，我用这个Workaround搞定了
  • 避坑指南：ZYNQ AXI_IIC读写EEPROM时常见的3个硬件陷阱
  • 15分钟解决Windows系统臃肿：WinUtil一站式优化工具全面评测
  • 低成本构建私有AI画室：雯雯的后宫-造相Z-Image-瑜伽女孩单机多模型并行部署方案
  • 如何快速完成重庆大学毕业论文格式排版？终极LaTeX模板使用指南
  • 龙芯k - 走马观碑组MPU驱动移植涎
  • SRWE窗口编辑器：打破Windows窗口限制的终极解决方案
  • 如何彻底解决Cursor试用限制：终极设备ID重置工具完整指南
  • Groq API+沉浸式翻译插件：5分钟搞定AI翻译神器（附详细配置截图）