零基础入门python24：分类 CRUD 与用户数据隔离

零基础入门python24分类 CRUD 与用户数据隔离一、上一篇课后练习讲解上一篇要求配置会话过期和 Cookie 安全属性。开发环境可以使用较短过期时间方便测试生产环境要启用SESSION_COOKIE_SECURE、HTTPOnly和合适的SameSite策略。上一篇课后练习完整答案上一篇练习的要求已落实到下面完整文件先运行项目测试再用 curl 对照状态码和数据库持久化结果答案要点登录用 check_password_hashuser_loader 从数据库加载/me 依赖 current_user未登录统一 401不暴露邮箱是否存在。文件app/auth.py完整参考答案文件完整文件app/auth.pyfromflaskimportBlueprint,requestfromflask_loginimportlogin_user,login_required,current_userfromwerkzeug.securityimportcheck_password_hashfrom.modelsimportUser bpBlueprint(auth,__name__,url_prefix/api/auth)bp.post(/login)deflogin():datarequest.get_json(silentTrue)or{}userUser.query.filter_by(emailstr(data.get(email,)).casefold()).first()ifnotuserornotcheck_password_hash(user.password_hash,str(data.get(password,))):return{error:invalid_credentials},401login_user(user)return{id:user.id,email:user.email}bp.get(/me)login_requireddefme():return{id:current_user.id,email:current_user.email}完整参考答案文件本篇对应的交付源码完整文件flask-ledger/app/auth.pyfromflaskimportBlueprint,requestfromflask_loginimportcurrent_user,login_required,login_user,logout_userfrom.extensionsimportdbfrom.modelsimportUser bpBlueprint(auth,__name__,url_prefix/api/auth)bp.post(/register)defregister():datarequest.get_json(silentTrue)or{}emailstr(data.get(email,)).strip().lower()passwordstr(data.get(password,))ifnotinemail:return{message:邮箱格式不正确},400iflen(password)8:return{message:密码至少8位},400ifdb.session.scalar(db.select(User).where(User.emailemail)):return{message:邮箱已注册},409userUser(emailemail)user.set_password(password)db.session.add(user)db.session.commit()return{id:user.id,email:user.email},201bp.post(/login)deflogin():datarequest.get_json(silentTrue)or{}emailstr(data.get(email,)).strip().lower()userdb.session.scalar(db.select(User).where(User.emailemail))ifnotuserornotuser.check_password(str(data.get(password,))):return{message:邮箱或密码错误},401login_user(user)return{id:user.id,email:user.email}bp.post(/logout)login_requireddeflogout():logout_user()return{message:已退出}bp.get(/me)login_requireddefme():return{id:current_user.id,email:current_user.email}验收命令python -m pytest -qDjango 项目使用 python manage.py test。预期测试通过若失败先检查迁移、配置和事务回滚。二、本篇完成什么分类是账目的前置资源。本篇实现当前用户的分类新增、列表、修改和删除并验证用户 A 不能读取或修改用户 B 的分类。三、为什么每个查询都要带 user_id只用Category.query.get(category_id)会让用户猜 id 后读到别人的数据。正确做法是把资源所有权放进查询条件where(Category.id id, Category.user_id current_user.id)。这样即使客户端传入别人的 id查询也不会返回记录。四、关键实现bp.post(/api/categories)login_requireddefcreate_category():datarequest.get_json(silentTrue)or{}namestr(data.get(name,)).strip()ifnotname:return{message:分类名不能为空},400ifdb.session.scalar(db.select(Category).where(Category.user_idcurrent_user.id,Category.namename)):return{message:分类已存在},409rowCategory(namename,user_idcurrent_user.id)db.session.add(row)db.session.commit()return{id:row.id,name:row.name},201重复检查是用户范围内的检查数据库联合唯一约束是最终防线两者同时存在前者提供友好错误后者应对并发请求。五、验收用户 A 登录后创建“餐饮”列表能看到用户 B 登录后列表为空修改 A 的分类返回 404。测试同时覆盖未登录 401、空名称 400、重复名称 409。课后练习增加分类删除接口并阻止删除仍有账目的分类。项目增量分类 CRUD 与用户隔离分类也是用户资源。查询、修改和删除都必须带 current_user.id只根据分类 id 查询会导致越权读取。categoryCategory.query.filter_by(idcategory_id,user_idcurrent_user.id).first_or_404()category.nameform.name.strip()db.session.commit()名称唯一约束应按 user_id name 组合而不是全站唯一。删除仍被账目使用的分类要返回业务错误或要求迁移账目不能静默丢失历史。验收与课后练习用户 A 不能读取、修改或删除用户 B 的分类重复名称返回 409课后增加分类删除保护测试。五、分类 CRUD 的完整闭环分类接口不是简单的四个 SQL。每一步都要带上当前用户条件避免用户 A 通过猜 id 访问用户 B 的数据ledger_bp.get(/categories)login_requireddeflist_categories():rows(Category.query.filter_by(user_idcurrent_user.id).order_by(Category.name.asc()).all())return{items:[{id:row.id,name:row.name}forrowinrows]}ledger_bp.post(/categories)login_requireddefcreate_category():name(request.get_json(silentTrue)or{}).get(name,).strip()ifnot1len(name)30:return{error:name_length},400rowCategory(user_idcurrent_user.id,namename)db.session.add(row)try:db.session.commit()exceptIntegrityError:db.session.rollback()return{error:category_exists},409return{id:row.id,name:row.name},201ledger_bp.delete(/categories/int:category_id)login_requireddefdelete_category(category_id):rowCategory.query.filter_by(idcategory_id,user_idcurrent_user.id).first()ifrowisNone:# 对不存在和不属于自己的资源统一返回 404避免泄露资源存在性。return{error:not_found},404db.session.delete(row)db.session.commit()return,204删除分类前要决定已有账目如何处理本项目把category_id设为可空删除时置空如果产品要求禁止删除就应该在服务层先统计引用数量并返回 409。不要依赖 SQLite 的默认外键行为因为不同数据库的外键开关不同。六、分页和搜索的边界可以用page、size两个参数并限制size 100pagemax(request.args.get(page,1,typeint),1)sizemin(max(request.args.get(size,20,typeint),1),100)keywordrequest.args.get(q,).strip()queryCategory.query.filter_by(user_idcurrent_user.id)ifkeyword:queryquery.filter(Category.name.ilike(f%{keyword}%))totalquery.count()itemsquery.order_by(Category.id.desc()).offset((page-1)*size).limit(size).all()count()和列表查询是两次 SQL数据量很大时可以改为游标分页现在先保证契约清晰。不要直接把size拼到 SQL 字符串里ORM 参数化会自动处理类型。七、上一篇练习讲解与错误排查上一篇的/auth/me要复用current_user不能从请求体接收 user id。测试至少覆盖A 创建分类后能看到B 看不到 A 的分类重复名称返回 409删除不存在的 id 返回 404。IntegrityError未回滚会让后续请求一直处于“事务已失败”状态表现为PendingRollbackError遇到该错误第一时间执行db.session.rollback()。八、本篇练习实现分类更新PUT /categories/id要求新名称仍在 1—30 字且不能与当前用户其他分类重复补充一个“引用账目时拒绝删除”的模式开关。下一篇将把分类 id 接入账目创建并讲 Decimal 与日期输入。本篇结束完整模块文件下面是交付项目中真实存在的完整文件 flask-ledger/app/ledger.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码复制单个函数会丢失上下文因此这里提供整份文件。fromdatetimeimportdatefromdecimalimportDecimal,InvalidOperationfromflaskimportBlueprint,requestfromflask_loginimportcurrent_user,login_requiredfromsqlalchemyimportfuncfrom.extensionsimportdbfrom.modelsimportCategory,Transaction bpBlueprint(ledger,__name__,url_prefix/api)defowned_category(category_id:int):returndb.session.scalar(db.select(Category).where(Category.idcategory_id,Category.user_idcurrent_user.id))bp.get(/categories)login_requireddeflist_categories():rowsdb.session.scalars(db.select(Category).where(Category.user_idcurrent_user.id).order_by(Category.name)).all()return[{id:row.id,name:row.name}forrowinrows]bp.post(/categories)login_requireddefcreate_category():namestr((request.get_json(silentTrue)or{}).get(name,)).strip()ifnot1len(name)40:return{message:分类名称长度应为1到40},400existsdb.session.scalar(db.select(Category).where(Category.user_idcurrent_user.id,Category.namename))ifexists:return{message:分类已存在},409rowCategory(namename,user_idcurrent_user.id)db.session.add(row)db.session.commit()return{id:row.id,name:row.name},201bp.post(/transactions)login_requireddefcreate_transaction():datarequest.get_json(silentTrue)or{}try:amountDecimal(str(data.get(amount,0))).quantize(Decimal(0.01))happened_ondate.fromisoformat(str(data.get(happened_on,date.today())))category_idint(data.get(category_id))except(InvalidOperation,ValueError,TypeError):return{message:金额、日期或分类格式不正确},400kindstr(data.get(kind,))ifkindnotin{income,expense}oramount0:return{message:类型或金额不正确},400ifnotowned_category(category_id):return{message:分类不存在},404rowTransaction(kindkind,amountamount,notestr(data.get(note,))[:200],happened_onhappened_on,category_idcategory_id,user_idcurrent_user.id,)db.session.add(row)db.session.commit()returnrow.to_dict(),201bp.get(/transactions)login_requireddeflist_transactions():pagemax(request.args.get(page,1,typeint),1)sizemin(max(request.args.get(size,10,typeint),1),100)querydb.select(Transaction).where(Transaction.user_idcurrent_user.id)ifkind:request.args.get(kind):queryquery.where(Transaction.kindkind)ifcategory_id:request.args.get(category_id,typeint):queryquery.where(Transaction.category_idcategory_id)rowsdb.session.scalars(query.order_by(Transaction.happened_on.desc(),Transaction.id.desc()).offset((page-1)*size).limit(size)).all()return{page:page,size:size,items:[row.to_dict()forrowinrows]}bp.delete(/transactions/int:transaction_id)login_requireddefdelete_transaction(transaction_id:int):rowdb.session.scalar(db.select(Transaction).where(Transaction.idtransaction_id,Transaction.user_idcurrent_user.id))ifnotrow:return{message:账目不存在},404db.session.delete(row)db.session.commit()return,204bp.get(/statistics/monthly)login_requireddefmonthly_statistics():monthrequest.args.get(month,date.today().strftime(%Y-%m))try:startdate.fromisoformat(month-01)exceptValueError:return{message:月份格式应为YYYY-MM},400enddate(start.year(start.month12),1ifstart.month12elsestart.month1,1)rowsdb.session.execute(db.select(Transaction.kind,func.sum(Transaction.amount)).where(Transaction.user_idcurrent_user.id,Transaction.happened_onstart,Transaction.happened_onend).group_by(Transaction.kind)).all()totals{income:Decimal(0),expense:Decimal(0)}totals.update({kind:Decimal(total)forkind,totalinrows})return{month:month,income:str(totals[income]),expense:str(totals[expense]),balance:str(totals[income]-totals[expense])}