用 FastAPI 写求职者登录注册这是一个仿 BOSS 直聘项目的后端模块技术栈是 FastAPI MySQL Redis 阿里云 OSS。下面按代码结构记录求职者登录注册这块怎么写。数据库表 models文件 app/models/job_seeker.py用 ORM 建表就是写一个 Python 类fromtortoise.modelsimportModelfromtortoiseimportfieldsclassJobSeeker(Model):idfields.IntField(pkTrue)mobilefields.CharField(max_length20,uniqueTrue)# 手机号不能重复passwordfields.CharField(max_length255,nullTrue)avatarfields.CharField(max_length500,nullTrue)# 存头像图片网址nicknamefields.CharField(max_length100,nullTrue)emailfields.CharField(max_length100,nullTrue)wechatfields.CharField(max_length100,nullTrue)create_timefields.DatetimeField(auto_now_addTrue)update_timefields.DatetimeField(auto_nowTrue)classMeta:tablet_job_seekeruniqueTrue 表示手机号不能重复nullTrue 表示可以为空auto_now_add 和 auto_now 让数据库自动记录注册时间和修改时间。参数校验 schemas文件 app/schemas/job_seeker.py前端传来的数据用 Pydantic 定规矩Field(…) 表示必传frompydanticimportBaseModel,FieldclassLoginMobileRequest(BaseModel):mobile:strField(...,title手机号)password:strField(...,title密码)classLoginSmsCodeRequest(BaseModel):mobile:strField(...)code:strField(...)classCreateUserRequest(BaseModel):mobile:strField(...)password:strField(...)code:strField(...)classUpdateUserRequest(BaseModel):mobile:strField(...)new_password:strField(...)code:strField(...)漏传的话 FastAPI 会自动报错不用自己处理。业务逻辑 services文件 app/services/job_seeker.py先准备两个工具importredis redis_clientredis.Redis(hostlocalhost,port6379,db6,decode_responsesTrue)importrandomdefget_random_code(length6):return.join(random.choices(1234567890,klength))手机号密码登录asyncdeflogin_by_mobile_password(req):jobseekerawaitJobSeeker.get_or_none(mobilereq.mobile)ifjobseekerisNone:raiseException(手机号不存在)ifjobseeker.password!req.password:raiseException(密码错误)access_token,refresh_tokencreate_tokens(str(jobseeker.id),jobseeker.mobile)return{jobseeker:jobseeker,access_token:access_token,refresh_token:refresh_token}这里密码是明文对比的练手可以正式项目要加密存储。发送登录验证码asyncdefsend_sms_code(mobile):jobseekerawaitJobSeeker.get_or_none(mobilemobile)ifjobseekerisNone:raiseException(手机号不存在)codeget_random_code(6)redis_client.set(fboss-api:login:sms:{mobile},code,ex120)returncode验证码存进 Redis120 秒后自动过期。return code 是为了方便测试正式项目应该调用短信服务发到手机。验证码登录asyncdeflogin_by_mobile_code(req):jobseekerawaitJobSeeker.get_or_none(mobilereq.mobile)ifjobseekerisNone:raiseException(手机号不存在)redis_coderedis_client.get(fboss-api:login:sms:{req.mobile})ifredis_codeisNone:raiseException(验证码已过期)ifredis_code!req.code:raiseException(验证码错误)access_token,refresh_tokencreate_tokens(str(jobseeker.id),jobseeker.mobile)redis_client.delete(fboss-api:login:sms:{req.mobile})return{jobseeker:jobseeker,access_token:access_token,refresh_token:refresh_token}先取 Redis 里的验证码过期或错误就报错通过后发 token 并删掉验证码。注册发注册验证码asyncdefsend_sms_code_create(mobile):codeget_random_code(6)redis_client.set(fboss-api:create:sms:{mobile},code,ex120)returncode正式建号asyncdefcreate_job(req):redis_coderedis_client.get(fboss-api:create:sms:{req.mobile})ifnotredis_code:raiseException(验证码已过期)ifredis_code!req.code:raiseException(验证码错误)ifawaitJobSeeker.get_or_none(mobilereq.mobile):raiseException(手机号已存在)new_jobseekerawaitJobSeeker.create(**dict(req))redis_client.delete(fboss-api:create:sms:{req.mobile})access_token,refresh_tokencreate_tokens(str(new_jobseeker.id),new_jobseeker.mobile)return{msg:注册成功,access_token:access_token,refresh_token:refresh_token}注册的 Redis key 用 boss-api:create:sms:登录用 boss-api:login:sms:两者分开互不干扰。JobSeeker.create(**dict(req)) 把前端参数直接变成一条新记录注册完直接发 token。忘记密码asyncdefforget_password(req):redis_coderedis_client.get(fboss-api:login:sms:{req.mobile})ifnotredis_code:raiseException(验证码已过期)ifredis_code!req.code:raiseException(验证码错误)jobseekerawaitJobSeeker.get_or_none(mobilereq.mobile)ifjobseekerisNone:raiseException(手机号不存在)jobseeker.passwordreq.new_passwordawaitjobseeker.save()# 改了一定要 save不然不会存进数据库redis_client.delete(fboss-api:login:sms:{req.mobile})return{msg:修改密码成功}上传头像asyncdefupdate_avatar(job_seeker_id,file):contentawaitfile.read()ossAliyunOSSTool()ok,resoss.upload_single_file(content,file.filename,oss_pathjob_seeker_avatar/)ifok:jobseekerawaitJobSeeker.get_or_none(idjob_seeker_id)jobseeker.avatarres[access_url]awaitjobseeker.save()returnresraiseException(上传失败)图片传到 OSSOSS 返回图片网址只把这个网址存进 avatar 字段。查看资料asyncdefget_job_seeker_info_by_id(job_seeker_id):jsawaitJobSeeker.get_or_none(idjob_seeker_id)js.nicknamef用户{js.mobile}ifjs.nicknameisNoneelsejs.nickname js.avatar默认头像网址ifjs.avatarisNoneelsejs.avatarreturnjs昵称和头像为空时给默认值。接口 apis文件 app/apis/job_seeker_api.pyservice 写好后包成接口网址job_seeker_router.post(/login_by_mobile_password)asyncdeflogin_by_mobile_password(req:LoginMobileRequest):resawaitJobSeekerService.login_by_mobile_password(req)return{code:1,message:登录成功,data:res}接口层只做三件事收参数、调 service、包成统一格式返回。返回格式 code 为 1 成功 0 失败message 是提示data 是数据。全部接口前缀 /job_seeker/login_by_mobile_password 密码登录/send_sms_code/{mobile} 发登录验证码/login_by_mobile_code 验证码登录/send_sms_code_create/{mobile} 发注册验证码/create_job 注册/forget_password 忘记密码/me 查看资料需登录/update_avatar 换头像需登录登录状态 JWT登录成功后发 token之后每次请求带上服务器验证就知道是谁。defcreate_tokens(user_id,username):payload{user_id:user_id,username:username,type:access}payload[exp]datetime.utcnow()timedelta(minutes120)access_tokenjwt.encode(payload,SECRET_KEY,algorithmHS256)rpayloadpayload.copy()rpayload[type]refreshrpayload[exp]datetime.utcnow()timedelta(days7)refresh_tokenjwt.encode(rpayload,SECRET_KEY,algorithmHS256)returnaccess_token,refresh_tokenaccess_token 有效期 2 小时每次请求都带refresh_token 有效期 7 天用来换新 token。app/core/depends.py 里从请求头取 token 并解析出 user_iddefget_job_seeker_info(authorization:strHeader(None)):ifnotauthorization:raiseException(请登录)tokenauthorization.replace(Bearer ,)payloadverify_token(token)ifpayloadisNone:raiseException(验证失败)returnpayload[user_id]需要登录的接口加 Depends(get_job_seeker_info)FastAPI 自动先验证并把 user_id 传进来job_seeker_router.get(/me)asyncdefget_me(job_seeker_idDepends(get_job_seeker_info)):jsawaitJobSeekerService.get_job_seeker_info_by_id(job_seeker_id)return{code:1,message:获取成功,data:js}统一异常处理service 里 raise 的异常由全局处理器接住转成统一返回defglobal_exception_handler(request,exc):returnJSONResponse(status_code500,content{code:0,message:str(exc)})在 main.py 注册app.add_exception_handler(Exception,global_exception_handler)小结把这块写完后端常用的建表、参数校验、分层、JWT 登录、Redis 存验证码、OSS 存图片、统一报错都接触到了。建议跟着敲一遍把遇到的报错自己解决掉。正式上线前还要注意密码加密、验证码真发短信、发码接口加限流。
