taolib.testing.oauth.services.account_service

taolib.testing.oauth.services.account_service#

OAuth 账户服务模块。

管理 OAuth 身份与本地用户账户的关联。

Classes#

OAuthAccountService

OAuth 账户服务。

Module Contents#

class taolib.testing.oauth.services.account_service.OAuthAccountService(connection_repo: taolib.testing.oauth.repository.connection_repo.OAuthConnectionRepository, activity_repo: taolib.testing.oauth.repository.activity_repo.OAuthActivityLogRepository, token_encryptor: taolib.testing.oauth.crypto.token_encryption.TokenEncryptor)#

OAuth 账户服务。

管理用户与 OAuth 提供商之间的关联关系。

参数:

  • connection_repo -- 连接仓储

  • activity_repo -- 活动日志仓储

  • token_encryptor -- Token 加密器

_connection_repo#
_activity_repo#
_token_encryptor#
async find_or_create_connection(user_info: taolib.testing.oauth.models.profile.OAuthUserInfo, token_data: dict[str, Any], ip_address: str = '', user_agent: str = '') → tuple[taolib.testing.oauth.models.connection.OAuthConnectionDocument, bool]#

查找或创建 OAuth 连接。

流程： 1. 按 (provider, provider_user_id) 查找已存在的连接 2. 如果找到，更新 Token 并返回 3. 如果未找到，创建新连接（状态为 PENDING_ONBOARDING）

参数:

  • user_info -- 标准化的用户信息

  • token_data -- Token 数据（含 access_token, refresh_token 等）

  • ip_address -- 客户端 IP

  • user_agent -- 客户端 User-Agent

返回:

(connection, is_new_user) 元组

将新的提供商关联到已有用户。

参数:

  • user_id -- 本地用户 ID

  • user_info -- 标准化的用户信息

  • token_data -- Token 数据

  • ip_address -- 客户端 IP

  • user_agent -- 客户端 User-Agent

返回:

新创建的连接文档

抛出:

OAuthAlreadyLinkedError -- 该提供商已关联到此用户

解除用户与提供商的关联。

确保至少保留一种认证方式。

参数:

  • user_id -- 用户 ID

  • provider -- 提供商名称

  • has_password -- 用户是否设置了密码

  • ip_address -- 客户端 IP

  • user_agent -- 客户端 User-Agent

返回:

是否成功解除关联

抛出:

OAuthCannotUnlinkError -- 无法解除最后一种认证方式

async get_user_connections(user_id: str) → list[taolib.testing.oauth.models.connection.OAuthConnectionResponse]#

获取用户的所有 OAuth 连接。

参数:

user_id -- 用户 ID

返回:

连接响应列表