discours.io/core
5
0
Fork 0
Code Pull Requests Actions Projects Activity

auth-fix
Some checks failed
Deploy on push / type-check (push) Failing after 6s
Details
Deploy on push / deploy (push) Has been skipped
Details

Browse Source
This commit is contained in:
Untone 2025-06-02 22:28:17 +03:00
parent 21d28a0d8b
commit 63c96ef965
2 changed files with 7 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
auth/tokens/base.py
View File

@ -28,13 +28,13 @@ class BaseTokenManager:
Returns: Returns:
str: Ключ токена str: Ключ токена
""" """
if token_type == TokenType.SESSION: if token_type == "session": # noqa: S105
return f"session:{identifier}:{token}" return f"session:{identifier}:{token}"
if token_type == TokenType.VERIFICATION: if token_type == "verification": # noqa: S105
return f"verification_token:{token}" return f"verification_token:{token}"
if token_type == TokenType.OAUTH_ACCESS: if token_type == "oauth_access": # noqa: S105
return f"oauth_access:{identifier}" return f"oauth_access:{identifier}"
if token_type == TokenType.OAUTH_REFRESH: if token_type == "oauth_refresh": # noqa: S105
return f"oauth_refresh:{identifier}" return f"oauth_refresh:{identifier}"
error_msg = f"Неизвестный тип токена: {token_type}" error_msg = f"Неизвестный тип токена: {token_type}"
@ -44,7 +44,7 @@ class BaseTokenManager:
@lru_cache(maxsize=500) @lru_cache(maxsize=500)
def _make_user_tokens_key(user_id: str, token_type: TokenType) -> str: def _make_user_tokens_key(user_id: str, token_type: TokenType) -> str:
"""Создает ключ для списка токенов пользователя""" """Создает ключ для списка токенов пользователя"""
if token_type == TokenType.SESSION: if token_type == "session": # noqa: S105
return f"user_sessions:{user_id}" return f"user_sessions:{user_id}"
return f"user_tokens:{user_id}:{token_type}" return f"user_tokens:{user_id}:{token_type}"

6
auth/tokens/oauth.py
View File

@ -81,10 +81,8 @@ class OAuthTokenManager(BaseTokenManager):
async def get_token(self, user_id: int, provider: str, token_type: TokenType) -> Optional[TokenData]: async def get_token(self, user_id: int, provider: str, token_type: TokenType) -> Optional[TokenData]:
"""Получает токен""" """Получает токен"""
if isinstance(token_type, TokenType): if token_type.startswith("oauth_"):
if token_type.startswith("oauth_"): return await self._get_oauth_data_optimized(token_type, str(user_id), provider)
return await self._get_oauth_data_optimized(token_type, str(user_id), provider) # type: ignore[arg-type]
return await self._get_token_data_optimized(token_type, str(user_id), provider) # type: ignore[arg-type]
return None return None
async def _get_oauth_data_optimized( async def _get_oauth_data_optimized(