企业官网建设流程全解析

零基础入门Python14｜SQLite与SQL CRUD：建立图书数据库

本篇图解：数据从哪里来、到哪里去

图中最重要的边界是事务：只有成功路径才提交数据，失败路径不能留下半条业务记录。

一、上一篇课后练习讲解

标签既可以作为任务的子资源，也可以作为筛选条件：

POST /api/v1/tasks/3/tags 请求体：{"name":"Python"} 成功：201 {"id":8,"name":"Python"} 错误：404 TASK_NOT_FOUND；409 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?tag=Python&page=1&size=20

删除URL同时包含任务编号和标签编号，可以校验这个标签确实属于该任务。按标签筛选仍属于任务列表查询，所以使用查询参数。

可执行验收答案

下面是一个不依赖 Web 框架的内存实现，完整体现 201、204、404、409 的判断：

classTagService:def__init__(self):self.tags:dict[int,str]={}self.links:set[tuple[int,int]]=set()self.next_id=1defadd(self,task_id:int,name:str)->tuple[int,dict]:iftask_id<=0ornotname.strip():return404,{"code":"TASK_NOT_FOUND"}ifany(t==namefortinself.tags.values()):return409,{"code":"TAG_ALREADY_EXISTS"}tag_id=self.next_id;self.next_id+=1self.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,Noneservice=TagService()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(price>0),stockINTEGERNOTNULLDEFAULT0CHECK(stock>=0));

主键唯一标识一行；NOT NULL禁止空值；UNIQUE保证ISBN不重复；CHECK保护价格和库存；DEFAULT为未提供库存时设0。业务代码校验提升用户体验，数据库约束负责最后一道保护。

四、CRUD四类SQL

INSERTINTObooks(isbn,title,price,stock)VALUES('978-1','Python入门',59.00,10);SELECTid,isbn,title,price,stockFROMbooksWHEREprice<100ORDERBYidDESC;UPDATEbooksSETstock=stock-1WHEREid=1ANDstock>0;DELETEFROMbooksWHEREid=1;

执行UPDATE或DELETE前，先用相同WHERE条件SELECT确认范围。忘记WHERE会影响整张表。

五、完整Python数据库程序

importsqlite3frompathlibimportPath DATABASE=Path("library.db")defconnect():connection=sqlite3.connect(DATABASE)connection.row_factory=sqlite3.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:cursor=connection.execute("INSERT INTO books (isbn, title, price, stock) VALUES (?, ?, ?, ?)",(isbn,title,price,stock),)returncursor.lastrowiddeflist_books(keyword=""):withconnect()asconnection:rows=connection.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):ifstock<0:raiseValueError("库存不能小于0")withconnect()asconnection:cursor=connection.execute("UPDATE books SET stock = ? WHERE id = ?",(stock,book_id),)ifcursor.rowcount==0:raiseValueError("图书不存在")defdelete_book(book_id):withconnect()asconnection:cursor=connection.execute("DELETE FROM books WHERE id = ?",(book_id,),)ifcursor.rowcount==0:raiseValueError("图书不存在")if__name__=="__main__":create_tables()try:book_id=add_book("978-1","Python入门",59.00,10)print("新增编号：",book_id)exceptsqlite3.IntegrityErroraserror:print("新增失败：",error)print(list_books("Python"))

with连接块正常结束时自动提交，发生异常时回滚。row_factory让每行可以按列名读取。

六、参数化查询防注入

错误写法：

sql=f"SELECT * FROM books WHERE title = '{keyword}'"

用户输入直接成为SQL结构。正确写法：

connection.execute("SELECT * FROM books WHERE title = ?",(keyword,),)

占位符只能用于值，不能替代表名和排序字段；动态排序必须使用白名单。

七、本篇验收

• 重复ISBN被数据库拒绝；
• 负价格和负库存被CHECK拒绝；
• 查询使用参数化；
• 修改删除检查rowcount；
• SQL异常不会留下半次提交；
• 重新运行程序能读取已保存图书。

八、课后练习

增加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','a@example.com');SELECTid,nameFROMusersWHEREnameLIKE'A%'ORDERBYidDESC;