taolib.testing.oauth.services.session_service#
OAuth 会话服务模块。
管理跨服务的 OAuth 会话。
Classes#
OAuth 会话服务。 |
Module Contents#
- class taolib.testing.oauth.services.session_service.OAuthSessionService(session_repo: taolib.testing.oauth.repository.session_repo.OAuthSessionRepository, redis_client, jwt_secret: str, jwt_algorithm: str = 'HS256', access_token_expire_minutes: int = 15, refresh_token_expire_days: int = 7)#
OAuth 会话服务。
创建和管理跨服务的 OAuth 会话，集成 JWT Token 生成。
- 参数:
session_repo -- 会话仓储
redis_client -- Redis 异步客户端
jwt_secret -- JWT 密钥
jwt_algorithm -- JWT 算法
access_token_expire_minutes -- Access Token 过期时间（分钟）
refresh_token_expire_days -- Refresh Token 过期时间（天）
- _session_repo#
- _redis#
- _jwt_secret#
- _jwt_algorithm = 'HS256'#
- _access_expire_min = 15#
- _refresh_expire_days = 7#
- _create_jwt(user_id: str, roles: list[str], token_type: str, expires_delta: datetime.timedelta) str#
创建 JWT Token。
使用与 config_center 兼容的 Token 格式。
- 参数:
user_id -- 用户 ID
roles -- 用户角色列表
token_type -- Token 类型 ("access" 或 "refresh")
expires_delta -- 过期时间增量
- 返回:
JWT Token 字符串
- async create_session(user_id: str, connection_id: str, provider: taolib.testing.oauth.models.enums.OAuthProvider | str, roles: list[str], ip_address: str = '', user_agent: str = '', session_ttl_hours: int = 24) dict#
创建新的 OAuth 会话。
- 参数:
user_id -- 用户 ID
connection_id -- OAuth 连接 ID
provider -- OAuth 提供商
roles -- 用户角色列表
ip_address -- 客户端 IP
user_agent -- 客户端 User-Agent
session_ttl_hours -- 会话有效期（小时）
- 返回:
包含 session_id、access_token、refresh_token 的字典
- async validate_session(session_id: str) taolib.testing.oauth.models.session.OAuthSessionDocument | None#
验证会话有效性。
优先从 Redis 缓存检查，回退到 MongoDB。
- 参数:
session_id -- 会话 ID
- 返回:
会话文档，无效则返回 None
- async refresh_session(session_id: str, roles: list[str]) dict | None#
刷新会话的 JWT Token。
- 参数:
session_id -- 会话 ID
roles -- 用户角色列表
- 返回:
新的 Token 字典，会话无效则返回 None
- async list_active_sessions(user_id: str) list[taolib.testing.oauth.models.session.OAuthSessionResponse]#
列出用户的所有活跃会话。
- 参数:
user_id -- 用户 ID
- 返回:
会话响应列表