用 FastAPI 接入钉钉第三方登录
在求职者登录注册之后，给项目加一个"用钉钉扫码登录"的功能。思路和普通登录不一样：用户不在我们这注册密码，而是去钉钉那边扫码授权，钉钉把用户信息回传给我们，我们再判断是否已绑定账号。
这套代码在 app/apis/third_part_api.py、app/services/third_party.py、app/schemas/third_party.py，另外用到了 job_seeker.py 里的 ThirdPart 模型。
第三方用户表 ThirdPart
文件 app/models/job_seeker.py
钉钉用户和求职者账号不是同一张表，单独存一张第三方表，再用外键关联到求职者：
class ThirdPart(models.Model):
id = fields.IntField(pk=True)
nick = fields.CharField(max_length=100, null=True)
avatar = fields.CharField(max_length=500, null=True)
openid = fields.CharField(max_length=100, null=True)
unionid = fields.CharField(max_length=100, null=True)
create_time = fields.DatetimeField(auto_now_add=True)
update_time = fields.DatetimeField(auto_now=True)
mobile = fields.CharField(max_length=20, null=True)
job_seeker = fields.ForeignKeyField(
"models.JobSeeker", null=True, related_name="third_part", on_delete=fields.CASCADE
)
class Meta:
table = "t_third_part"
openid 是钉钉给每个用户的唯一标识，后面就用它来判断"这个人之前登录过没有"。job_seeker 这个外键指向求职者表，为空就表示还没绑定我们的账号。
参数校验 schemas
文件 app/schemas/third_party.py
绑定手机号时前端要传手机号和验证码：
from pydantic import BaseModel, Field
class BindMobileToJobSeeker(BaseModel):
mobile: str = Field(..., title="手机号", example="13200000000")
code: str = Field(..., title="验证码", example="123456")
业务逻辑 services
文件 app/services/third_party.py
redis_client = redis.Redis(
host="localhost", port=6379, db=6, decode_responses=True, protocol=2
)
判断是否已绑定 thrid_party_is_bind
钉钉回调拿到用户信息后，先走这个函数：
@staticmethod
async def thrid_party_is_bind(user_info: dict):
redis_client.set(
f"boss-api:bind:dingding:{user_info['mobile']}",
json.dumps(user_info, ensure_ascii=False),
ex=60 * 5
)
thridpart = await ThirdPart.get_or_none(openid=user_info["openId"])
if not thridpart:
thridpart = await ThirdPart.create(
openid=user_info["openId"],
nick=user_info["nick"],
avatar=user_info["avatarUrl"],
mobile=user_info["mobile"],
unionid=user_info["unionId"]
)
return False, None, None
thridpart.mobile = user_info["mobile"]
thridpart.avatar = user_info["avatarUrl"]
thridpart.nick = user_info["nick"]
thridpart.unionid = user_info["unionId"]
thridpart.openid = user_info["openId"]
await thridpart.save()
job_seeker = await thridpart.fetch_related("job_seeker")
if not thridpart.job_seeker:
return False, None, None
job_seeker_id = thridpart.job_seeker.id
access_token, refresh_token = create_tokens(job_seeker_id, user_info["mobile"])
return True, access_token, refresh_token
流程是：先把用户信息存进 Redis（5 分钟过期，后面绑定手机号要用），然后按 openid 查第三方表。查不到说明第一次来，建一条第三方记录，返回 False 让前端去绑定页。查到了就更新信息，再看看有没有关联的求职者账号。有就直接发 token 登录；没有就返回 False 去绑定。这里返回值用 is_bind 这个布尔告诉前端"要不要跳绑定页"。
绑定手机号 bind_mobile_to_job_seeker
@staticmethod
async def bind_mobile_to_job_seeker(bindMobileToJobSeekerRequest: BindMobileToJobSeeker):
key = f"boss-api:bind:sms:{bindMobileToJobSeekerRequest.mobile}"
redis_code = redis_client.get(key)
if redis_code is None:
raise Exception("验证码已过期")
if redis_code != bindMobileToJobSeekerRequest.code:
raise Exception("验证码错误")
job_seeker = await JobSeeker.get_or_none(mobile=bindMobileToJobSeekerRequest.mobile)
ding_user_info = redis_client.get(f"boss-api:bind:dingding:{bindMobileToJobSeekerRequest.mobile}")
ding_user_info_dict = json.loads(ding_user_info)
async with in_transaction() as conn:
if job_seeker is None:
job_seeker = await JobSeeker.create(
mobile=bindMobileToJobSeekerRequest.mobile,
nickname=ding_user_info_dict["nick"],
avatar=ding_user_info_dict["avatarUrl"],
using_db=conn
)
thridparty = await ThirdPart.get_or_none(openid=ding_user_info_dict["openId"])
thridparty.job_seeker = job_seeker
await thridparty.save(using_db=conn)
access_token, refresh_token = create_tokens(job_seeker.id, job_seeker.mobile)
return access_token, refresh_token, job_seeker.id
先校验验证码，再从 Redis 取出之前存的钉钉用户信息。然后开一个事务：手机号在我们这没账号就新建一个（昵称头像直接用钉钉的），再把第三方记录的外键 job_seeker 指向这个新账号。这样以后再用钉钉登录就能直接找到账号了。用事务是为了保证"建账号"和"绑定"要么都成功要么都失败，不会只绑一半。
接口 apis
文件 app/apis/third_part_api.py
钉钉回调
@third_party_router.get("/dingtalk/login/callback")
async def dingtalk_callback(code: str):
code_access_token_dict = {
"clientId": "dingz9b12c4wuh7lfs2i",
"clientSecret": "HxKE8M-…省略",
"code": code,
"grantType": "authorization_code"
}
token_res = httpx.post(
url="https://api.dingtalk.com/v1.0/oauth2/userAccessToken",
json=code_access_token_dict
)
accessToken = token_res.json()["accessToken"]
user_res = httpx.get(
url="https://api.dingtalk.com/v1.0/contact/users/me",
headers={"x-acs-dingtalk-access-token": accessToken}
)
user_info = {
"nick": user_res.json()["nick"],
"avatarUrl": user_res.json()["avatarUrl"],
"mobile": user_res.json()["mobile"],
"unionId": user_res.json()["unionId"],
"openId": user_res.json()["openId"]
}
is_bind, access_token, refresh_token = await ThirdPartyService.thrid_party_is_bind(user_info)
html_content = f"""
<html><body><script>
var messageData = {{
type: 'dingtalk_callback',
token: '{access_token}',
refresh_token: '{refresh_token}',
is_bind: {json.dumps(is_bind)},
userInfo: {json.dumps(user_info)}
}};
window.opener.postMessage(messageData, '*');
setTimeout(function() {{ window.close(); }}, 500);
</script></body></html>
"""
return HTMLResponse(content=html_content)
这是整个流程的入口。前端调起钉钉扫码，用户扫完钉钉带着一个 code 跳回这个回调地址。我们拿着 code 加 clientId、clientSecret 去换 accessToken，再用 accessToken 调接口拿到用户昵称、头像、手机号、openId。接着调刚才的 thrid_party_is_bind 判断是否已绑定，最后返回一个 HTML 页面，用 postMessage 把结果发给打开扫码窗口的父页面，再自动关掉自己。
绑定手机号接口
@third_party_router.post("/bind_mobile_to_job_seeker")
async def bind_mobile_to_job_seeker(bindMobileToJobSeekerRequest: BindMobileToJobSeeker):
access_token, refresh_token, job_seeker_id = await ThirdPartyService.bind_mobile_to_job_seeker(
bindMobileToJobSeekerRequest
)
return {
"code": 1,
"msg": "success",
"data": {
"access_token": access_token,
"refresh_token": refresh_token,
"job_seeker_id": job_seeker_id
}
}
前端在绑定页让用户输入手机号和验证码，调这个接口完成绑定并拿到 token。
完整流程
小结
第三方登录的核心就是把"在我们这注册"换成"在钉钉授权，我们拿用户信息"，再用一张第三方表记录对应关系，最后靠绑定手机号把第三方账号和我们的求职者账号连起来。这里用到的 Redis 临时存用户信息、事务保证绑定完整、postMessage 回传结果，都是比较实用的写法。
几个可以留意的点：回调里 httpx.post 和 httpx.get 是同步写法，放在 async 函数里会阻塞事件循环，正式项目换成 httpx.AsyncClient 更好；clientSecret 直接写在代码里，应该放到配置文件或环境变量；绑定验证码的 Redis key 是 boss-api:bind:sms:，记得发码的地方也要用同样的 key。
