用 FastAPI 写求职者登录注册

这是一个仿 BOSS 直聘项目的后端模块，技术栈是 FastAPI + MySQL + Redis + 阿里云 OSS。下面按代码结构记录求职者登录注册这块怎么写。

数据库表 models

文件 app/models/job_seeker.py

用 ORM 建表就是写一个 Python 类：

fromtortoise.modelsimportModelfromtortoiseimportfieldsclassJobSeeker(Model):id=fields.IntField(pk=True)mobile=fields.CharField(max_length=20,unique=True)# 手机号，不能重复password=fields.CharField(max_length=255,null=True)avatar=fields.CharField(max_length=500,null=True)# 存头像图片网址nickname=fields.CharField(max_length=100,null=True)email=fields.CharField(max_length=100,null=True)wechat=fields.CharField(max_length=100,null=True)create_time=fields.DatetimeField(auto_now_add=True)update_time=fields.DatetimeField(auto_now=True)classMeta:table="t_job_seeker"

unique=True 表示手机号不能重复，null=True 表示可以为空，auto_now_add 和 auto_now 让数据库自动记录注册时间和修改时间。

参数校验 schemas

文件 app/schemas/job_seeker.py

前端传来的数据用 Pydantic 定规矩，Field(…) 表示必传：

frompydanticimportBaseModel,FieldclassLoginMobileRequest(BaseModel):mobile:str=Field(...,title="手机号")password:str=Field(...,title="密码")classLoginSmsCodeRequest(BaseModel):mobile:str=Field(...)code:str=Field(...)classCreateUserRequest(BaseModel):mobile:str=Field(...)password:str=Field(...)code:str=Field(...)classUpdateUserRequest(BaseModel):mobile:str=Field(...)new_password:str=Field(...)code:str=Field(...)

漏传的话 FastAPI 会自动报错，不用自己处理。

业务逻辑 services

文件 app/services/job_seeker.py

先准备两个工具：

importredis redis_client=redis.Redis(host="localhost",port=6379,db=6,decode_responses=True)importrandomdefget_random_code(length=6):return''.join(random.choices('1234567890',k=length))

手机号密码登录

asyncdeflogin_by_mobile_password(req):jobseeker=awaitJobSeeker.get_or_none(mobile=req.mobile)ifjobseekerisNone:raiseException("手机号不存在")ifjobseeker.password!=req.password:raiseException("密码错误")access_token,refresh_token=create_tokens(str(jobseeker.id),jobseeker.mobile)return{"jobseeker":jobseeker,"access_token":access_token,"refresh_token":refresh_token}

这里密码是明文对比的，练手可以，正式项目要加密存储。

发送登录验证码

asyncdefsend_sms_code(mobile):jobseeker=awaitJobSeeker.get_or_none(mobile=mobile)ifjobseekerisNone:raiseException("手机号不存在")code=get_random_code(6)redis_client.set(f"boss-api:login:sms:{mobile}",code,ex=120)returncode

验证码存进 Redis，120 秒后自动过期。return code 是为了方便测试，正式项目应该调用短信服务发到手机。

验证码登录

asyncdeflogin_by_mobile_code(req):jobseeker=awaitJobSeeker.get_or_none(mobile=req.mobile)ifjobseekerisNone:raiseException("手机号不存在")redis_code=redis_client.get(f"boss-api:login:sms:{req.mobile}")ifredis_codeisNone:raiseException("验证码已过期")ifredis_code!=req.code:raiseException("验证码错误")access_token,refresh_token=create_tokens(str(jobseeker.id),jobseeker.mobile)redis_client.delete(f"boss-api:login:sms:{req.mobile}")return{"jobseeker":jobseeker,"access_token":access_token,"refresh_token":refresh_token}

先取 Redis 里的验证码，过期或错误就报错，通过后发 token 并删掉验证码。

注册

发注册验证码：

asyncdefsend_sms_code_create(mobile):code=get_random_code(6)redis_client.set(f"boss-api:create:sms:{mobile}",code,ex=120)returncode

正式建号：

asyncdefcreate_job(req):redis_code=redis_client.get(f"boss-api:create:sms:{req.mobile}")ifnotredis_code:raiseException("验证码已过期")ifredis_code!=req.code:raiseException("验证码错误")ifawaitJobSeeker.get_or_none(mobile=req.mobile):raiseException("手机号已存在")new_jobseeker=awaitJobSeeker.create(**dict(req))redis_client.delete(f"boss-api:create:sms:{req.mobile}")access_token,refresh_token=create_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_code=redis_client.get(f"boss-api:login:sms:{req.mobile}")ifnotredis_code:raiseException("验证码已过期")ifredis_code!=req.code:raiseException("验证码错误")jobseeker=awaitJobSeeker.get_or_none(mobile=req.mobile)ifjobseekerisNone:raiseException("手机号不存在")jobseeker.password=req.new_passwordawaitjobseeker.save()# 改了一定要 save，不然不会存进数据库redis_client.delete(f"boss-api:login:sms:{req.mobile}")return{"msg":"修改密码成功"}

上传头像

asyncdefupdate_avatar(job_seeker_id,file):content=awaitfile.read()oss=AliyunOSSTool()ok,res=oss.upload_single_file(content,file.filename,oss_path="job_seeker_avatar/")ifok:jobseeker=awaitJobSeeker.get_or_none(id=job_seeker_id)jobseeker.avatar=res["access_url"]awaitjobseeker.save()returnresraiseException("上传失败")

图片传到 OSS，OSS 返回图片网址，只把这个网址存进 avatar 字段。

查看资料

asyncdefget_job_seeker_info_by_id(job_seeker_id):js=awaitJobSeeker.get_or_none(id=job_seeker_id)js.nickname=f"用户{js.mobile}"ifjs.nicknameisNoneelsejs.nickname js.avatar="默认头像网址"ifjs.avatarisNoneelsejs.avatarreturnjs

昵称和头像为空时给默认值。

接口 apis

文件 app/apis/job_seeker_api.py

service 写好后包成接口网址：

@job_seeker_router.post("/login_by_mobile_password")asyncdeflogin_by_mobile_password(req:LoginMobileRequest):res=awaitJobSeekerService.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(minutes=120)access_token=jwt.encode(payload,SECRET_KEY,algorithm="HS256")rpayload=payload.copy()rpayload["type"]="refresh"rpayload["exp"]=datetime.utcnow()+timedelta(days=7)refresh_token=jwt.encode(rpayload,SECRET_KEY,algorithm="HS256")returnaccess_token,refresh_token

access_token 有效期 2 小时，每次请求都带；refresh_token 有效期 7 天，用来换新 token。

app/core/depends.py 里从请求头取 token 并解析出 user_id：

defget_job_seeker_info(authorization:str=Header(None)):ifnotauthorization:raiseException("请登录")token=authorization.replace("Bearer ","")payload=verify_token(token)ifpayloadisNone:raiseException("验证失败")returnpayload["user_id"]

需要登录的接口加 Depends(get_job_seeker_info)，FastAPI 自动先验证并把 user_id 传进来：

@job_seeker_router.get("/me")asyncdefget_me(job_seeker_id=Depends(get_job_seeker_info)):js=awaitJobSeekerService.get_job_seeker_info_by_id(job_seeker_id)return{"code":1,"message":"获取成功","data":js}

统一异常处理

service 里 raise 的异常由全局处理器接住，转成统一返回：

defglobal_exception_handler(request,exc):returnJSONResponse(status_code=500,content={"code":0,"message":str(exc)})

在 main.py 注册：

app.add_exception_handler(Exception,global_exception_handler)

小结

把这块写完，后端常用的建表、参数校验、分层、JWT 登录、Redis 存验证码、OSS 存图片、统一报错都接触到了。建议跟着敲一遍，把遇到的报错自己解决掉。正式上线前还要注意密码加密、验证码真发短信、发码接口加限流。