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关联两个表的字段SQLStructured Query Language用于操作数据库的语言常用的数据库操作库1. sqlite3sqlite3是 Python 内置的 SQLite 数据库接口用于操作 SQLite 数据库。importsqlite3# 连接数据库connsqlite3.connect(example.db)# 创建游标cconn.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. psycopg2psycopg2是 PostgreSQL 数据库的 Python 接口用于操作 PostgreSQL 数据库。importpsycopg2# 连接数据库connpsycopg2.connect(hostlocalhost,databasetest,userpostgres,passwordpassword)# 创建游标curconn.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. SQLAlchemySQLAlchemy是一个功能强大的 ORM对象关系映射库用于操作各种数据库。fromsqlalchemyimportcreate_engine,Column,Integer,Stringfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmaker# 创建引擎enginecreate_engine(sqlite:///example.db)# 创建基类Basedeclarative_base()# 定义模型classUser(Base):__tablename__usersidColumn(Integer,primary_keyTrue)nameColumn(String)ageColumn(Integer)# 创建表Base.metadata.create_all(engine)# 创建会话Sessionsessionmaker(bindengine)sessionSession()# 插入数据user1User(nameAlice,age25)user2User(nameBob,age30)session.add_all([user1,user2])session.commit()# 查询数据userssession.query(User).all()foruserinusers:print(fID:{user.id}, Name:{user.name}, Age:{user.age})# 更新数据usersession.query(User).filter_by(nameAlice).first()user.age26session.commit()# 删除数据usersession.query(User).filter_by(nameBob).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_namedb_name self.connNoneself.cursorNoneself.connect()self.create_tables()defconnect(self):self.connsqlite3.connect(self.db_name)self.cursorself.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_passwordself.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):userself.get_user(username)ifuser:hashed_passwordself.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_idNone):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__:dbDatabase(user_management.db)# 添加用户db.add_user(admin,password123,adminexample.com)db.add_user(user1,password456,user1example.com)# 验证用户print(db.verify_password(admin,password123))# Trueprint(db.verify_password(admin,wrongpassword))# False# 添加帖子userdb.get_user(admin)ifuser:db.add_post(user[0],First Post,Hello World!)db.add_post(user[0],Second Post,How are you?)# 获取帖子postsdb.get_posts()forpostinposts:print(fID:{post[0]}, User ID:{post[1]}, Title:{post[2]}, Content:{post[3]}, Created At:{post[4]})db.close()数据库操作的最佳实践1. 使用参数化查询# 不好的做法cursor.execute(fSELECT * FROM users WHERE username {username})# 好的做法cursor.execute(SELECT * FROM users WHERE username ?,(username,))2. 关闭数据库连接# 使用上下文管理器withsqlite3.connect(example.db)asconn:cursorconn.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()raise4. 索引优化-- 创建索引CREATEINDEXidx_users_usernameONusers(username);CREATEINDEXidx_posts_user_idONposts(user_id);5. 错误处理try:# 执行数据库操作exceptsqlite3.Errorase:print(fDatabase error:{e})exceptExceptionase:print(fError:{e})finally:# 关闭连接ifconn:conn.close()常见问题与解决方案1. 数据库连接失败问题无法连接到数据库出现连接错误。解决方案检查数据库服务器是否运行检查连接参数是否正确检查网络连接是否正常检查数据库权限是否正确2. 数据插入失败问题数据插入失败出现完整性错误。解决方案检查数据是否符合表结构要求检查唯一约束是否被违反检查外键约束是否被违反使用参数化查询避免 SQL 注入3. 查询性能问题问题查询速度慢影响应用性能。解决方案创建适当的索引优化查询语句减少查询返回的数据量使用分页查询考虑使用缓存4. 事务处理问题问题事务处理不当导致数据不一致。解决方案使用 try-except-finally 结构处理事务确保在异常情况下回滚事务避免长时间持有事务考虑使用分布式事务对于复杂系统5. 安全问题问题数据库操作存在安全风险如 SQL 注入。解决方案使用参数化查询对用户输入进行验证和过滤限制数据库用户的权限加密敏感数据定期备份数据库总结Python 数据库操作是后端开发的重要组成部分它可以帮助我们存储和管理数据。通过本文的学习我们了解了数据库的基本概念、常用的数据库操作库、实战案例、最佳实践和常见问题与解决方案。作为一个从后端转 Rust 的萌新我认为学习 Python 的数据库操作是非常有价值的。它不仅可以帮助我们与数据库进行交互还可以让我们更好地理解数据存储和管理的原理。在进行数据库操作时我们应该使用参数化查询、关闭数据库连接、使用事务、优化索引和处理错误。同时我们还应该注意解决数据库连接失败、数据插入失败、查询性能问题、事务处理问题和安全问题等常见问题。保持学习保持输出今天的 Python 数据库操作实战文章就到这里希望对大家有所帮助。欢迎在评论区分享你的经验和问题我们一起进步