零基础入门Python14SQLite与SQL CRUD建立图书数据库本篇图解数据从哪里来、到哪里去图中最重要的边界是事务只有成功路径才提交数据失败路径不能留下半条业务记录。一、上一篇课后练习讲解标签既可以作为任务的子资源也可以作为筛选条件POST /api/v1/tasks/3/tags 请求体{name:Python} 成功201 {id:8,name:Python} 错误404 TASK_NOT_FOUND409 TAG_ALREADY_EXISTS GET /api/v1/tasks/3/tags 成功200 {items:[...]} DELETE /api/v1/tasks/3/tags/8 成功204 错误404 TAG_NOT_FOUND GET /api/v1/tasks?tagPythonpage1size20删除URL同时包含任务编号和标签编号可以校验这个标签确实属于该任务。按标签筛选仍属于任务列表查询所以使用查询参数。可执行验收答案下面是一个不依赖 Web 框架的内存实现完整体现 201、204、404、409 的判断classTagService:def__init__(self):self.tags:dict[int,str]{}self.links:set[tuple[int,int]]set()self.next_id1defadd(self,task_id:int,name:str)-tuple[int,dict]:iftask_id0ornotname.strip():return404,{code:TASK_NOT_FOUND}ifany(tnamefortinself.tags.values()):return409,{code:TAG_ALREADY_EXISTS}tag_idself.next_id;self.next_id1self.tags[tag_id]name;self.links.add((task_id,tag_id))return201,{id:tag_id,name:name}defdelete(self,task_id:int,tag_id:int)-tuple[int,dict|None]:if(task_id,tag_id)notinself.links:return404,{code:TAG_NOT_FOUND}self.links.remove((task_id,tag_id));self.tags.pop(tag_id,None)return204,NoneserviceTagService()assertservice.add(3,Python)[0]201assertservice.add(3,Python)[0]409assertservice.delete(3,1)[0]204assertservice.delete(3,1)[0]404真实 SQLite 实现把links换成task_tags(task_id,tag_id)表和唯一索引状态码由路由层返回业务层仍应返回明确结果对象。二、本篇成果使用Python内置sqlite3创建图书数据库学习表、行、列、主键、约束和参数化SQL完成新增、查询、修改和删除并解释为什么不能拼接用户输入。三、表结构CREATETABLEbooks(idINTEGERPRIMARYKEYAUTOINCREMENT,isbnTEXTNOTNULLUNIQUE,titleTEXTNOTNULL,priceNUMERICNOTNULLCHECK(price0),stockINTEGERNOTNULLDEFAULT0CHECK(stock0));主键唯一标识一行NOT NULL禁止空值UNIQUE保证ISBN不重复CHECK保护价格和库存DEFAULT为未提供库存时设0。业务代码校验提升用户体验数据库约束负责最后一道保护。四、CRUD四类SQLINSERTINTObooks(isbn,title,price,stock)VALUES(978-1,Python入门,59.00,10);SELECTid,isbn,title,price,stockFROMbooksWHEREprice100ORDERBYidDESC;UPDATEbooksSETstockstock-1WHEREid1ANDstock0;DELETEFROMbooksWHEREid1;执行UPDATE或DELETE前先用相同WHERE条件SELECT确认范围。忘记WHERE会影响整张表。五、完整Python数据库程序importsqlite3frompathlibimportPath DATABASEPath(library.db)defconnect():connectionsqlite3.connect(DATABASE)connection.row_factorysqlite3.Row connection.execute(PRAGMA foreign_keys ON)returnconnectiondefcreate_tables():withconnect()asconnection:connection.execute( CREATE TABLE IF NOT EXISTS books ( id INTEGER PRIMARY KEY AUTOINCREMENT, isbn TEXT NOT NULL UNIQUE, title TEXT NOT NULL, price NUMERIC NOT NULL CHECK (price 0), stock INTEGER NOT NULL DEFAULT 0 CHECK (stock 0) ) )defadd_book(isbn,title,price,stock):withconnect()asconnection:cursorconnection.execute(INSERT INTO books (isbn, title, price, stock) VALUES (?, ?, ?, ?),(isbn,title,price,stock),)returncursor.lastrowiddeflist_books(keyword):withconnect()asconnection:rowsconnection.execute(SELECT id, isbn, title, price, stock FROM books WHERE title LIKE ? ORDER BY id DESC,(f%{keyword}%,),).fetchall()return[dict(row)forrowinrows]defupdate_stock(book_id,stock):ifstock0:raiseValueError(库存不能小于0)withconnect()asconnection:cursorconnection.execute(UPDATE books SET stock ? WHERE id ?,(stock,book_id),)ifcursor.rowcount0:raiseValueError(图书不存在)defdelete_book(book_id):withconnect()asconnection:cursorconnection.execute(DELETE FROM books WHERE id ?,(book_id,),)ifcursor.rowcount0:raiseValueError(图书不存在)if__name____main__:create_tables()try:book_idadd_book(978-1,Python入门,59.00,10)print(新增编号,book_id)exceptsqlite3.IntegrityErroraserror:print(新增失败,error)print(list_books(Python))with连接块正常结束时自动提交发生异常时回滚。row_factory让每行可以按列名读取。六、参数化查询防注入错误写法sqlfSELECT * FROM books WHERE title {keyword}用户输入直接成为SQL结构。正确写法connection.execute(SELECT * FROM books WHERE title ?,(keyword,),)占位符只能用于值不能替代表名和排序字段动态排序必须使用白名单。七、本篇验收重复ISBN被数据库拒绝负价格和负库存被CHECK拒绝查询使用参数化修改删除检查rowcountSQL异常不会留下半次提交重新运行程序能读取已保存图书。八、课后练习增加authors表和book_authors中间表让一本书可有多位作者、作者可写多本书写出建表SQL和参数化Python函数add_author_to_book。下一篇会讲JOIN、聚合、索引和事务。实战补充SQLite 事务实验完成任务和写审计属于一个用例任一失败都应该回滚。withsqlite3.connect(tasks.db)asdb:db.execute(UPDATE tasks SET done 1 WHERE id ?,(task_id,))db.execute(INSERT INTO audit(action, subject_id) VALUES (?, ?),(task.done,task_id))故意让第二条 SQL 违反约束重新查询第一条更新应不存在。课后练习增加外键和 CHECK记录约束错误如何转换成 409。本篇结束完整模块文件本节不是代码片段而是本篇结束时该模块的完整版本。请先备份旧文件再整体替换替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理未涉及的代码先不要自行删减。本篇完整示例CREATETABLEusers(idINTEGERPRIMARYKEY,nameTEXTNOTNULL,emailTEXTUNIQUE);INSERTINTOusers(name,email)VALUES(Alice,aexample.com);SELECTid,nameFROMusersWHEREnameLIKEA%ORDERBYidDESC; 零基础入门Python14｜SQLite与SQL CRUD：建立图书数据库
零基础入门Python14SQLite与SQL CRUD建立图书数据库本篇图解数据从哪里来、到哪里去图中最重要的边界是事务只有成功路径才提交数据失败路径不能留下半条业务记录。一、上一篇课后练习讲解标签既可以作为任务的子资源也可以作为筛选条件POST /api/v1/tasks/3/tags 请求体{name:Python} 成功201 {id:8,name:Python} 错误404 TASK_NOT_FOUND409 TAG_ALREADY_EXISTS GET /api/v1/tasks/3/tags 成功200 {items:[...]} DELETE /api/v1/tasks/3/tags/8 成功204 错误404 TAG_NOT_FOUND GET /api/v1/tasks?tagPythonpage1size20删除URL同时包含任务编号和标签编号可以校验这个标签确实属于该任务。按标签筛选仍属于任务列表查询所以使用查询参数。可执行验收答案下面是一个不依赖 Web 框架的内存实现完整体现 201、204、404、409 的判断classTagService:def__init__(self):self.tags:dict[int,str]{}self.links:set[tuple[int,int]]set()self.next_id1defadd(self,task_id:int,name:str)-tuple[int,dict]:iftask_id0ornotname.strip():return404,{code:TASK_NOT_FOUND}ifany(tnamefortinself.tags.values()):return409,{code:TAG_ALREADY_EXISTS}tag_idself.next_id;self.next_id1self.tags[tag_id]name;self.links.add((task_id,tag_id))return201,{id:tag_id,name:name}defdelete(self,task_id:int,tag_id:int)-tuple[int,dict|None]:if(task_id,tag_id)notinself.links:return404,{code:TAG_NOT_FOUND}self.links.remove((task_id,tag_id));self.tags.pop(tag_id,None)return204,NoneserviceTagService()assertservice.add(3,Python)[0]201assertservice.add(3,Python)[0]409assertservice.delete(3,1)[0]204assertservice.delete(3,1)[0]404真实 SQLite 实现把links换成task_tags(task_id,tag_id)表和唯一索引状态码由路由层返回业务层仍应返回明确结果对象。二、本篇成果使用Python内置sqlite3创建图书数据库学习表、行、列、主键、约束和参数化SQL完成新增、查询、修改和删除并解释为什么不能拼接用户输入。三、表结构CREATETABLEbooks(idINTEGERPRIMARYKEYAUTOINCREMENT,isbnTEXTNOTNULLUNIQUE,titleTEXTNOTNULL,priceNUMERICNOTNULLCHECK(price0),stockINTEGERNOTNULLDEFAULT0CHECK(stock0));主键唯一标识一行NOT NULL禁止空值UNIQUE保证ISBN不重复CHECK保护价格和库存DEFAULT为未提供库存时设0。业务代码校验提升用户体验数据库约束负责最后一道保护。四、CRUD四类SQLINSERTINTObooks(isbn,title,price,stock)VALUES(978-1,Python入门,59.00,10);SELECTid,isbn,title,price,stockFROMbooksWHEREprice100ORDERBYidDESC;UPDATEbooksSETstockstock-1WHEREid1ANDstock0;DELETEFROMbooksWHEREid1;执行UPDATE或DELETE前先用相同WHERE条件SELECT确认范围。忘记WHERE会影响整张表。五、完整Python数据库程序importsqlite3frompathlibimportPath DATABASEPath(library.db)defconnect():connectionsqlite3.connect(DATABASE)connection.row_factorysqlite3.Row connection.execute(PRAGMA foreign_keys ON)returnconnectiondefcreate_tables():withconnect()asconnection:connection.execute( CREATE TABLE IF NOT EXISTS books ( id INTEGER PRIMARY KEY AUTOINCREMENT, isbn TEXT NOT NULL UNIQUE, title TEXT NOT NULL, price NUMERIC NOT NULL CHECK (price 0), stock INTEGER NOT NULL DEFAULT 0 CHECK (stock 0) ) )defadd_book(isbn,title,price,stock):withconnect()asconnection:cursorconnection.execute(INSERT INTO books (isbn, title, price, stock) VALUES (?, ?, ?, ?),(isbn,title,price,stock),)returncursor.lastrowiddeflist_books(keyword):withconnect()asconnection:rowsconnection.execute(SELECT id, isbn, title, price, stock FROM books WHERE title LIKE ? ORDER BY id DESC,(f%{keyword}%,),).fetchall()return[dict(row)forrowinrows]defupdate_stock(book_id,stock):ifstock0:raiseValueError(库存不能小于0)withconnect()asconnection:cursorconnection.execute(UPDATE books SET stock ? WHERE id ?,(stock,book_id),)ifcursor.rowcount0:raiseValueError(图书不存在)defdelete_book(book_id):withconnect()asconnection:cursorconnection.execute(DELETE FROM books WHERE id ?,(book_id,),)ifcursor.rowcount0:raiseValueError(图书不存在)if__name____main__:create_tables()try:book_idadd_book(978-1,Python入门,59.00,10)print(新增编号,book_id)exceptsqlite3.IntegrityErroraserror:print(新增失败,error)print(list_books(Python))with连接块正常结束时自动提交发生异常时回滚。row_factory让每行可以按列名读取。六、参数化查询防注入错误写法sqlfSELECT * FROM books WHERE title {keyword}用户输入直接成为SQL结构。正确写法connection.execute(SELECT * FROM books WHERE title ?,(keyword,),)占位符只能用于值不能替代表名和排序字段动态排序必须使用白名单。七、本篇验收重复ISBN被数据库拒绝负价格和负库存被CHECK拒绝查询使用参数化修改删除检查rowcountSQL异常不会留下半次提交重新运行程序能读取已保存图书。八、课后练习增加authors表和book_authors中间表让一本书可有多位作者、作者可写多本书写出建表SQL和参数化Python函数add_author_to_book。下一篇会讲JOIN、聚合、索引和事务。实战补充SQLite 事务实验完成任务和写审计属于一个用例任一失败都应该回滚。withsqlite3.connect(tasks.db)asdb:db.execute(UPDATE tasks SET done 1 WHERE id ?,(task_id,))db.execute(INSERT INTO audit(action, subject_id) VALUES (?, ?),(task.done,task_id))故意让第二条 SQL 违反约束重新查询第一条更新应不存在。课后练习增加外键和 CHECK记录约束错误如何转换成 409。本篇结束完整模块文件本节不是代码片段而是本篇结束时该模块的完整版本。请先备份旧文件再整体替换替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理未涉及的代码先不要自行删减。本篇完整示例CREATETABLEusers(idINTEGERPRIMARYKEY,nameTEXTNOTNULL,emailTEXTUNIQUE);INSERTINTOusers(name,email)VALUES(Alice,aexample.com);SELECTid,nameFROMusersWHEREnameLIKEA%ORDERBYidDESC;