1. 前言

在现代Web应用中，第三方登录已成为提升用户体验、简化注册流程的重要功能。钉钉作为国内广泛使用的企业办公平台，其OAuth2.0登录能力为应用提供了便捷的用户认证方案。本文将详细介绍如何在FastAPI框架中实现钉钉第三方登录功能。

2. 准备工作

2.1 申请钉钉开发者账号

首先需要访问钉钉开放平台，使用企业钉钉账号登录。如果没有企业账号，可以创建测试企业或使用个人开发者模式。

2.2 创建企业内部应用

1. 登录钉钉开放平台控制台
2. 点击"应用开发" → "企业内部开发" → "H5微应用"
3. 点击"创建应用"，填写应用基本信息：
   • 应用名称：FastAPI钉钉登录示例
   • 应用图标：上传合适的图标
   • 应用描述：FastAPI集成钉钉登录的演示应用
4. 创建完成后，记录以下关键信息：
   • AppKey：应用的唯一标识
   • AppSecret：用于获取access_token的密钥
   • AgentId：应用ID（可选）

3. OAuth2.0授权流程原理

钉钉第三方登录基于OAuth2.0授权码模式，主要流程如下：

1. 用户点击"钉钉登录"按钮
2. 应用重定向到钉钉授权页面
3. 用户扫码或输入账号密码授权
4. 钉钉回调到应用指定的redirect_uri，携带授权码
5. 应用使用授权码换取access_token
6. 使用access_token获取用户信息

4. 构建授权URL

根据钉钉开放平台文档，构建授权URL的格式如下：

# 钉钉OAuth2.0授权URL模板
AUTH_URL_TEMPLATE = "https://login.dingtalk.com/oauth2/auth"
构建授权URL参数
def build_auth_url(app_key: str, redirect_uri: str, state: str = None) -> str:
params = {
"redirect_uri": redirect_uri,
"response_type": "code",
"client_id": app_key,
"scope": "openid corpid",
"prompt": "consent"
}
if state:
    params["state"] = state
URL编码并拼接
query_string = "&".join([f"{k}={urllib.parse.quote(v)}" for k, v in params.items()])
return f"{AUTH_URL_TEMPLATE}?{query_string}"</code></pre>
实际生成的授权URL示例：
https://login.dingtalk.com/oauth2/auth?
redirect_uri=http://127.0.0.1:8000/third_party/dingtalk/login/callback
&response_type=code
&client_id=dingmbiza7wx7lrsl05z   # 应用的AppKey
&scope=openid corpid  # 此处的openid保持不变
&state=random_state_string
&prompt=consent
5. FastAPI后端实现
5.1 项目结构
fastapi-dingtalk-auth/
├── app/
│   ├── init.py
│   ├── main.py
│   ├── config.py
│   ├── core/
│   │   ├── init.py
│   │   ├── logging.py
│   │   └── security.py
│   ├── api/
│   │   ├── init.py
│   │   └── v1/
│   │       ├── init.py
│   │       └── auth.py
│   └── services/
│       ├── init.py
│       └── dingtalk.py
├── requirements.txt
└── .env
5.2 配置文件
app/config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
钉钉配置
DINGTALK_APP_KEY: str
DINGTALK_APP_SECRET: str
DINGTALK_REDIRECT_URI: str = "http://127.0.0.1:8000/api/v1/auth/dingtalk/callback"
服务器配置
HOST: str = "127.0.0.1"
PORT: int = 8000
DEBUG: bool = True
class Config:
env_file = ".env"
settings = Settings()
5.3 钉钉服务层
app/services/dingtalk.py
import httpx
from typing import Optional, Dict, Any
from app.config import settings
from app.core.logging import logger
class DingTalkService:
def init(self):
self.app_key = settings.DINGTALK_APP_KEY
self.app_secret = settings.DINGTALK_APP_SECRET
self.redirect_uri = settings.DINGTALK_REDIRECT_URI
self.base_url = "https://oapi.dingtalk.com"
async def get_access_token(self, auth_code: str) -> Optional[Dict[str, Any]]:
"""使用授权码获取access_token"""
url = f"{self.base_url}/sns/gettoken"
params = {
"appid": self.app_key,
"appsecret": self.app_secret
}
try:
async with httpx.AsyncClient() as client:
response = await client.get(url, params=params)
response.raise_for_status()
result = response.json()
    if result.get("errcode") == 0:
        access_token = result.get("access_token")
        # 使用access_token和auth_code获取用户信息
        return await self.get_user_info(access_token, auth_code)
    else:
        logger.error(f"获取access_token失败: {result}")
        return None
except Exception as e:
logger.error(f"获取access_token异常: {e}")
return None
async def get_user_info(self, access_token: str, auth_code: str) -> Optional[Dict[str, Any]]:
"""获取用户信息"""
url = f"{self.base_url}/sns/getuserinfo_bycode"
params = {
"access_token": access_token
}
data = {
"tmp_auth_code": auth_code
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(url, params=params, json=data)
response.raise_for_status()
result = response.json()
    if result.get("errcode") == 0:
        user_info = result.get("user_info", {})
        return {
            "unionid": user_info.get("unionid"),
            "openid": user_info.get("openid"),
            "nick": user_info.get("nick"),
            "avatar": user_info.get("avatar"),
            "corpid": user_info.get("corpid")
        }
    else:
        logger.error(f"获取用户信息失败: {result}")
        return None
except Exception as e:
logger.error(f"获取用户信息异常: {e}")
return None
def generate_auth_url(self, state: str = None) -> str:
"""生成钉钉授权URL"""
import urllib.parse
params = {
"redirect_uri": self.redirect_uri,
"response_type": "code",
"client_id": self.app_key,
"scope": "openid corpid",
"prompt": "consent"
}
if state:
params["state"] = state
query_string = "&amp;".join([f"{k}={urllib.parse.quote(v)}" for k, v in params.items()])
return f"https://login.dingtalk.com/oauth2/auth?{query_string}"&lt;/code&gt;&lt;/pre&gt;
5.4 认证路由
app/api/v1/auth.py
from fastapi import APIRouter, Depends, Request, HTTPException
from fastapi.responses import RedirectResponse
from typing import Optional
import secrets
from app.services.dingtalk import DingTalkService
from app.core.logging import logger
router = APIRouter(prefix="/auth", tags=["认证"])
初始化钉钉服务
dingtalk_service = DingTalkService()
@router.get("/dingtalk/login")
async def dingtalk_login(state: Optional[str] = None):
"""
跳转到钉钉登录页面
"""
生成随机的state参数，用于防止CSRF攻击
if not state:
state = secrets.token_urlsafe(16)
auth_url = dingtalk_service.generate_auth_url(state)
logger.info(f"生成钉钉授权URL: {auth_url}")
return RedirectResponse(url=auth_url)
@router.get("/dingtalk/callback")
async def dingtalk_callback(
request: Request,
code: Optional[str] = None,
authCode: Optional[str] = None,
state: Optional[str] = None
):
"""
钉钉授权回调接口
"""
logger.info(f"收到钉钉回调 - code: {code}, authCode: {authCode}, state: {state}")
验证必要参数
if not authCode and not code:
raise HTTPException(status_code=400, detail="缺少授权码参数")
使用authCode或code（钉钉不同版本参数名可能不同）
auth_code = authCode or code
try:
获取用户信息
user_info = await dingtalk_service.get_access_token(auth_code)
if not user_info:
raise HTTPException(status_code=401, detail="获取用户信息失败")
logger.info(f"用户登录成功: {user_info}")
这里可以：
1. 创建或更新本地用户记录
2. 生成JWT token返回给前端
3. 设置session或cookie
示例：返回用户信息和自定义token
return {
"code": 0,
"message": "登录成功",
"data": {
"user_info": user_info,
"access_token": "your_jwt_token_here",  # 实际应生成JWT
"token_type": "bearer",
"expires_in": 7200
}
}
except Exception as e:
logger.error(f"钉钉登录处理异常: {e}")
raise HTTPException(status_code=500, detail="登录处理失败")
@router.get("/user/profile")
async def get_user_profile():
"""
获取当前用户信息（需要认证）
"""
实际应从JWT或session中获取用户ID
return {
"code": 0,
"message": "success",
"data": {
"user_id": "user_123",
"nickname": "测试用户",
"avatar": "https://example.com/avatar.jpg",
"source": "dingtalk"
}
}
6. 前端集成示例
6.1 HTML登录按钮
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>FastAPI钉钉登录示例</title>
<style>
.login-container {
max-width: 400px;
margin: 100px auto;
padding: 30px;
border: 1px solid #ddd;
border-radius: 8px;
text-align: center;
}
.dingtalk-btn {
background-color: #0086FF;
color: white;
border: none;
padding: 12px 24px;
border-radius: 4px;
font-size: 16px;
cursor: pointer;
display: flex;
align-items: center;
justify-content: center;
gap: 10px;
margin: 20px auto;
}
.dingtalk-btn:hover {
background-color: #0070D9;
}
.dingtalk-icon {
width: 20px;
height: 20px;
}
&lt;/style&gt;
</head>
<body>
<div class="login-container">
<h2>欢迎登录</h2>
<p>请选择登录方式</p>
&lt;button class="dingtalk-btn" onclick="dingtalkLogin()"&gt;
&lt;img src="https://img.alicdn.com/imgextra/i4/O1CN01hOw5yQ1b8hZ9QY8U5_!!6000000003415-2-tps-200-200.png"
alt="钉钉" class="dingtalk-icon"&gt;
钉钉登录
&lt;/button&gt;
&amp;lt;div id="user-info" style="display: none;"&amp;gt;
&amp;lt;h3&amp;gt;登录成功&amp;lt;/h3&amp;gt;
&amp;lt;p id="user-name"&amp;gt;&amp;lt;/p&amp;gt;
&amp;lt;img id="user-avatar" src="" alt="头像" style="width: 80px; height: 80px; border-radius: 50%;"&amp;gt;
&amp;lt;/div&amp;gt;
&lt;/div&gt;
&lt;script&gt;
function dingtalkLogin() {
// 跳转到后端授权接口
window.location.href = 'http://127.0.0.1:8000/api/v1/auth/dingtalk/login';
}
// 检查URL中是否有回调参数
function checkCallback() {
const urlParams = new URLSearchParams(window.location.search);
const code = urlParams.get('code');
const authCode = urlParams.get('authCode');
if (code || authCode) {
    // 如果有回调参数，发送到后端处理
    fetch(`/api/v1/auth/dingtalk/callback?code=${code || ''}&amp;amp;authCode=${authCode || ''}`)
        .then(response =&amp;gt; response.json())
        .then(data =&amp;gt; {
            if (data.code === 0) {
                document.getElementById('user-name').textContent = data.data.user_info.nick;
                document.getElementById('user-avatar').src = data.data.user_info.avatar;
                document.getElementById('user-info').style.display = 'block';
            // 存储token
            localStorage.setItem('access_token', data.data.access_token);
        }
    });
}
}
// 页面加载时检查
window.onload = checkCallback;
&lt;/script&gt;
</body>
</html>
7. 环境配置与运行
7.1 安装依赖
requirements.txt
fastapi==0.104.1
uvicorn[standard]==0.24.0
httpx==0.25.1
python-dotenv==1.0.0
pydantic-settings==2.1.0
pip install -r requirements.txt
7.2 环境变量配置
.env 文件
DINGTALK_APP_KEY=your_app_key_here
DINGTALK_APP_SECRET=your_app_secret_here
DINGTALK_REDIRECT_URI=http://127.0.0.1:8000/api/v1/auth/dingtalk/callback
7.3 启动应用
启动开发服务器
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
8. 常见问题与调试
8.1 常见错误码
错误码
说明
解决方案
40001
无效的AppKey
检查AppKey是否正确，应用是否已启用
40002
无效的AppSecret
检查AppSecret是否正确，是否已重置
40003
无效的授权码
检查authCode是否过期或已被使用
40004
redirect_uri不匹配
检查回调地址是否与应用配置一致
8.2 调试技巧