1) Рефакторинг.

2) Авторизация через телеграм.

Author: ProFastCode

app/api/deps.py

@@ -2,13 +2,14 @@
 Dependencies
 """
 
-from fastapi import Depends, Header
+from fastapi import Depends
+from fastapi.security import HTTPAuthorizationCredentials
 from typing_extensions import Annotated
 
 from app import models
 from app.core import exps, settings
 from app.core.db import Database, SessionLocal
-from app.core.security import JWTManager, TelegramAuth
+from app.core.security import JWTManager, oauth
 
 
 async def get_db() -> Database:
@@ -20,18 +21,18 @@ async def get_jwt_manager() -> JWTManager:
     return JWTManager(settings.APP_SECRET_KEY)
 
 
-async def get_telegram_auth() -> TelegramAuth:
-    return TelegramAuth(
+async def get_oauth_telegram() -> oauth.Telegram:
+    return oauth.Telegram(
         settings.TELEGRAM_BOT_TOKEN, settings.TELEGRAM_BOT_USERNAME
     )
 
 
 async def get_current_user(
-        access_token: Annotated[str, Header()],
-        db: Annotated[Database, Depends(get_db)],
-        jwt_manager: Annotated[JWTManager, Depends(get_jwt_manager)],
+    credentials: Annotated[HTTPAuthorizationCredentials, oauth.oauth],
+    jwt_manager: Annotated[JWTManager, Depends(get_jwt_manager)],
+    db: Annotated[Database, Depends(get_db)],
 ) -> models.User:
-    payload = jwt_manager.decode_token(access_token)
+    payload = jwt_manager.decode_token(credentials.credentials)
     if payload.get('type') != 'access':
         raise exps.TOKEN_INVALID
     if not (user := await db.user.read(payload.get('id'))):

app/api/endpoints/tokens/auth.py

@@ -9,24 +9,24 @@
 from app.api import deps
 from app.core import exps
 from app.core.db import Database
-from app.core.security import JWTManager, TelegramAuth
+from app.core.security import JWTManager, oauth
 
 router = APIRouter(prefix='/auth')
 
 
 @router.post('/telegram/', response_model=models.AuthToken)
 async def telegram(
-        user: models.UserCreate,
-        db: Annotated[Database, Depends(deps.get_db)],
-        auth: Annotated[TelegramAuth, Depends(deps.get_telegram_auth)],
-        jwt_manager: Annotated[JWTManager, Depends(deps.get_jwt_manager)],
+    user: models.UserCreate,
+    db: Annotated[Database, Depends(deps.get_db)],
+    jwt_manager: Annotated[JWTManager, Depends(deps.get_jwt_manager)],
+    oauth_telegram: Annotated[oauth, Depends(deps.get_oauth_telegram)],
 ):
     """
     Get auth token
     """
     user_dict = user.model_dump()
-    computed_hash = auth.generate_hash(user_dict)
-    if not auth.is_correct(computed_hash, user.hash):
+    computed_hash = oauth_telegram.generate_hash(user_dict)
+    if not oauth_telegram.is_correct(computed_hash, user.hash):
         raise exps.USER_IS_CORRECT
 
     model_user = models.User(**user.model_dump())
@@ -35,7 +35,7 @@ async def telegram(
     else:
         await db.user.create(model_user)
     await db.session.commit()
-    token = models.AuthToken(token=jwt_manager.encode_token(
-        {'id': user.id, 'type': 'auth'},15
-    ))
+    token = models.AuthToken(
+        token=jwt_manager.encode_token({'id': user.id, 'type': 'auth'}, 15)
+    )
     return token

app/api/endpoints/tokens/pair.py

@@ -22,8 +22,12 @@ async def new_pair_tokens(
     if payload.get('type') != 'auth':
         raise exps.TOKEN_INVALID
     payload['type'] = 'access'
-    access_token = models.AccessToken(token=jwt_manager.encode_token(payload, 120))
+    access_token = models.AccessToken(
+        token=jwt_manager.encode_token(payload, 120)
+    )
     payload['type'] = 'refresh'
-    refresh_token = models.RefreshToken(token=jwt_manager.encode_token(payload, 1440))
+    refresh_token = models.RefreshToken(
+        token=jwt_manager.encode_token(payload, 1440)
+    )
     tokens = models.PairTokens(access=access_token, refresh=refresh_token)
     return tokens

app/core/security/__init__.py

@@ -1,4 +1,4 @@
-from .auth import TelegramAuth
+from . import oauth
 from .jwt_manager import JWTManager
 
-__all__ = ['JWTManager', 'TelegramAuth']
+__all__ = ['JWTManager', 'oauth']

app/core/security/auth/__init__.py

@@ -24,11 +24,10 @@ def decode_token(self, token: str) -> dict:
 
         return payload
 
-    def encode_token(self, payload: dict, minutes: int | None = None) -> str:
-        claims = {'payload': payload}
-        if minutes:
-            claims['exp'] = dt.datetime.now(dt.UTC) + dt.timedelta(
-                minutes=minutes
-            )
+    def encode_token(self, payload: dict, minutes: int) -> str:
+        claims = {
+            'payload': payload,
+            'exp': dt.datetime.now(dt.UTC) + dt.timedelta(minutes=minutes)
+        }
         token = jwt.encode(claims, self.secret_key, algorithm=ALGORITHMS.HS256)
         return token

app/core/security/jwt_manager/jwt_manager.py renamed to app/core/security/jwt_manager.py

@@ -1,15 +1,13 @@
 import hashlib
 import hmac
 
-from app.core import settings
+from fastapi.security import HTTPBearer
 
+oauth = HTTPBearer()
 
-class TelegramAuth:
-    def __init__(
-        self,
-        token: str = settings.TELEGRAM_BOT_TOKEN,
-        username: str = settings.TELEGRAM_BOT_USERNAME,
-    ):
+
+class Telegram:
+    def __init__(self, token: str, username: str):
         self.token = token
         self.username = username