跳到主要内容

第一个带认证的连接器

我们在对接三方系统的时候，一般情况下三方系统都需要进行身份认证，并且认证信息在后续的接口调用中需要使用。因此连接器实现时，需要有一个专门的模块来维护身份认证过程。

接下来我们以连接天智的FastBI系统为例来介绍一下如何实现身份认证。

1：确定认证方案​

在开始编码之前，首先需要先研究你要对接的第三方服务的认证机制：

  1. 查阅第三方服务的API文档，了解其认证方式
  2. 确定需要用户提供哪些信息（用户名、密码、API Key等）
  3. 确定认证流程（Basic认证、OAuth2、自定义Token等）

常见的认证类型：

  • Basic/Digest认证 ：最简单的HTTP认证方式，直接使用用户名和密码
  • Custom认证 ：自定义认证流程，常用于Token机制
  • OAuth2认证 ：标准的OAuth2流程，适用于多数现代Web服务

天智FastBI系统采用的是Token机制，因此我们使用Custom认证模式

2：创建认证配置​

你需要在连接器中创建一个认证配置字典，这个字典告诉系统如何处理认证：

auth = {
    'test': auth_test,
    'fields': [
        # 认证字段列表
    ],
    'auth_type': 'custom',
    'config_custom': {
        'implement': token_auth,
    },
}

配置项详解：​

  1. test ：指向测试认证是否成功的函数
  2. fields ：用户需要填写的认证信息字段列表
  3. auth_type ：认证类型（'basic', 'digest', 'custom', 'oauth2'）
  4. config_custom ：仅当auth_type为'custom'时需要，包含认证实现函数

3：定义认证字段​

在FastBI系统中，获取token需要url username password三个字段，认证字段定义如下：

'fields': [
    {
        'key': 'url',
        'required': True,
        'label': 'URL地址',
        'help_text': '登录地址',
    },
    {
        'key': 'username',
        'required': True,
        'label': '用户名',
        'help_text': '登录用户名',
    },
    {
        'key': 'password',
        'required': True,
        'label': '密码',
        'help_text': '登录密码',
    },
]

字段属性说明：​

  • key （必需）：字段的唯一标识符，后续代码中通过这个key获取字段值
  • required （必需）：布尔值，表示该字段是否必须填写
  • label （必需）：显示给用户的字段名称
  • help_text （可选）：对字段的进一步说明

更多的属性参考字段定义章节

4：实现认证函数​

示例中的认证函数token_auth负责执行实际的认证过程，在这个过程中调用了FastBI系统的登录接口，获取到了token。该函数返回一个字典，这些数据将被缓存在系统中，后续其它动作可通过入参auth_data取得这些信息：

def token_auth(auth_data:dict, *args, **kwargs):
    username: str = auth_data['username']
    password: str = auth_data['password']
    url: str = auth_data['url']
    if not url.lower().startswith('http'):
        url = 'http://' + url
    url = urlparse.urljoin(url, f'/api/rest/user/login/')
    response: Response = requests.request(method='POST', url=url, data={"username": username, "password": password})
    token = response.json()['token']
    return {"token": 'token ' + str(token)}

实现要点：​

  1. 函数接收的第一个参数auth_data是包含用户输入的所有认证字段的字典
  2. 通过字段的key从auth_data中获取对应值
  3. 执行认证逻辑（向认证服务器发送POST请求获取token）
  4. 返回包含认证信息的字典，供后续API调用使用

5：实现测试函数​

测试函数会在前端点击授权测试的时候被触发，本例中的测试函数auth_test用于验证认证是否成功，先从入参auth_data里获取token并调用个人信息接口，如果接口调用成功则表示测试通过，需要返回True：

def auth_test(*args, ** kwargs):
    auth_data = kwargs['auth_data']
    token: str = auth_data['token']
    url: str = auth_data['url']

    headers = {'Authorization': token}

    if not url.lower().startswith('http'):
        url = 'http://' + url
    # 发送请求
    url = urlparse.urljoin(url, f'/api/rest/user/me/')
    response: Response = requests.request(method='GET', url=url, headers=headers)
    res: dict = response.json()
    if response.status_code == 200:
        return True
    else:
        raise Exception('授权失败！(API Token输入错误)')

实现要点：​

  1. 从kwargs['auth_data']中获取认证函数返回的认证信息
  2. 使用这些信息发送一个测试请求（获取用户信息）
  3. 根据响应判断认证是否有效
  4. 有效返回True，无效抛出异常

6：整合认证配置到API定义​

将认证配置添加到你的API定义中：

def api():
    auth = {
        'test': auth_test,
        'fields': [
            {
                'key': 'url',
                'required': True,
                'label': 'URL地址',
                'help_text': '登录地址',
            },
            {
                'key': 'username',
                'required': True,
                'label': '用户名',
                'help_text': '登录用户名',
            },
            {
                'key': 'password',
                'required': True,
                'label': '密码',
                'help_text': '登录密码',
            },
        ],
        'auth_type': 'custom',
        'config_custom': {
            'implement': token_auth,
        },
    }

    api_dict = {
        'key': 'FastBITest',
        'actions': {
            # 动作定义
        },
        'label': 'FastBITest',
        'description': '商业智能可视化数据分析工具',
        "image": "https://fbi.tizdata.com/flow/FastBI.svg",
        'help_url': 'https://www.yuque.com/lixuegao/yae7w2/kk1hprf8zg7bv4s7',
        'authentication': auth,
        'category': '数据分析',
    }

    return api_dict

7：注意事项和最佳实践​

错误处理​

  1. 始终处理网络异常和HTTP错误状态码
  2. 提供清晰易懂的错误信息给用户
  3. 对异常情况进行适当处理

安全性​

  1. 不要在日志或错误信息中暴露敏感信息（如密码、token）
  2. 确保使用HTTPS进行认证和API调用
  3. 合理存储和传输认证信息

用户体验​

  1. 提供清晰的字段标签和帮助文本
  2. 对必填字段进行明确标识
  3. 在测试函数中提供有用的反馈信息

代码质量​

  1. 添加适当的注释说明复杂逻辑
  2. 遵循一致的命名规范
  3. 保持函数简洁，单一职责

总结​

实现连接器认证功能的核心步骤：

  1. 研究第三方服务 ：了解其认证机制和所需信息
  2. 设计认证字段 ：定义用户需要填写的信息
  3. 选择认证类型 ：根据服务特点选择合适的认证方式
  4. 实现认证函数 ：编写获取认证信息的代码
  5. 实现测试函数 ：编写验证认证有效性的代码
  6. 整合到API定义 ：将认证配置加入整体API结构
  7. 关注细节 ：注意错误处理、安全性、用户体验等方面

按照这些步骤，你应该能够为任何第三方服务实现连接器认证功能。记住，关键是理解目标服务的认证机制，并据此实现相应的代码逻辑。FastBI示例展示了自定义认证的完整实现过程，你可以参考这个模式来实现自己的连接器认证功能。