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

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

Author: ProFastCode

.env.dist

@@ -9,3 +9,5 @@ POSTGRES_PORT=
 POSTGRES_USER=
 POSTGRES_PASSWORD=
 POSTGRES_DATABASE=
+
+TELEGRAM_BOT_TOKEN=

Makefile

@@ -9,14 +9,25 @@ help:
 	@echo "  migrate		Alembic migrate database"
 	@echo "  generate		Alembic generate database"
 
+
+.PHONY:	blue
+blue:
+	poetry run blue app/
+
+.PHONY: isort
+isort:
+	poetry run isort app/
+
+.PHONY: ruff
+ruff:
+	poetry run ruff check app/ --fix --respect-gitignore
+
 .PHONY: ref
-ref:
-	poetry run ruff ./app --fix && poetry run black ./app
+ref: blue isort ruff
 
 .PHONY: dev
 dev:
-	uvicorn app:app --reload --access-log --log-level debug
-
+	poetry run fastapi dev app
 
 .PHONY: generate
 generate:

app/__init__.py

@@ -11,7 +11,7 @@
     title=settings.APP_TITLE,
     version=settings.APP_VERSION,
     description=settings.APP_DESCRIPTION,
-    openapi_url=f"{settings.APP_API_PREFIX}/openapi.json"
+    openapi_url=f'{settings.APP_API_PREFIX}/openapi.json',
 )
 
 app.include_router(api.api_router, prefix=settings.APP_API_PREFIX)

app/__main__.py

@@ -6,5 +6,5 @@
 
 from app import app
 
-if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+if __name__ == '__main__':
+    uvicorn.run(app, host='0.0.0.0', port=8000)

app/api/__init__.py

@@ -4,8 +4,7 @@
 
 from fastapi import APIRouter
 
-from .endpoints import users, tokens
+from . import endpoints
 
 api_router = APIRouter()
-api_router.include_router(users.router)
-api_router.include_router(tokens.router)
+api_router.include_router(endpoints.router)

app/api/deps.py

@@ -2,35 +2,32 @@
 Dependencies
 """
 
-from fastapi import Depends, HTTPException, status, Header
+from fastapi import Depends, Header
+from typing_extensions import Annotated
 
 from app import models
-from app.core import security
-from app.core.db import SessionLocal, Database
+from app.core import exps, settings
+from app.core.db import Database, SessionLocal
+from app.core.security import JWTTokenManager
 
 
 async def get_db() -> Database:
     async with SessionLocal() as session:
         yield Database(session)
 
 
+async def get_tkn_manager() -> JWTTokenManager:
+    return JWTTokenManager(settings.APP_SECRET_KEY)
+
+
 async def get_current_user(
-        short_token: str = Header(),
-        db: Database = Depends(get_db),
+    access_token: Annotated[str, Header()],
+    db: Annotated[Database, Depends(get_db)],
+    tkn_manager: Annotated[JWTTokenManager, Depends(get_tkn_manager)],
 ) -> models.User:
-    payload = security.tkn_manager.decode_short_token(short_token)
-    if not (user := await db.user.read(payload.get("id"))):
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found"
-        )
+    payload = tkn_manager.decode_token(access_token)
+    if payload.get('type') != 'access':
+        raise exps.TOKEN_INVALID
+    if not (user := await db.user.read(payload.get('id'))):
+        raise exps.USER_NOT_FOUND
     return user
-
-
-async def is_superuser(
-        user: models.User = Depends(get_current_user),
-) -> None:
-    if not user.is_superuser:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="You do not have access to this section",
-        )

app/api/endpoints/__init__.py

@@ -0,0 +1,11 @@
+"""
+Endpoints API Module
+"""
+
+from fastapi import APIRouter
+
+from . import tokens, users
+
+router = APIRouter()
+router.include_router(users.router)
+router.include_router(tokens.router)

app/api/endpoints/tokens/__init__.py

@@ -5,9 +5,8 @@
 from fastapi import APIRouter
 
 from . import auth, pair, refresh
-from app.core.structures import Tags
 
-router = APIRouter(prefix="/tokens", tags=[Tags.tokens])
+router = APIRouter(prefix='/tokens', tags=['tokens'])
 router.include_router(auth.router)
 router.include_router(pair.router)
 router.include_router(refresh.router)

app/api/endpoints/tokens/auth.py

@@ -1,24 +1,50 @@
+"""
+User Endpoints Module
+"""
+import hmac
+
 from fastapi import APIRouter, Depends
+from typing_extensions import Annotated
 
 from app import models
 from app.api import deps
-from app.core import exps
+from app.core import exps, settings
 from app.core.db import Database
-from app.core.security import tkn_manager, pwd_manager
+from app.core.security import JWTTokenManager
 
 router = APIRouter()
 
 
-@router.post("/auth/", response_model=models.AuthToken)
-async def new_auth_token(data: models.UserCreate, db: Database = Depends(deps.get_db)):
+@router.post('/auth/', response_model=models.AuthToken)
+async def auth(
+        user: models.UserCreate,
+        db: Annotated[Database, Depends(deps.get_db)],
+        tkn_manager: Annotated[JWTTokenManager, Depends(deps.get_tkn_manager)],
+):
     """
-    Получить токен аутентификации
+    Get auth token
     """
-    if not (user := await db.user.get_by_email(data.email)):
-        raise exps.USER_NOT_REGISTERED
-
-    if not pwd_manager.verify_password(data.password, user.password):
-        raise exps.USER_INCORRECT_PASSWORD
+    data_check_string = '\n'.join(
+        sorted(
+            f'{x}={y}'
+            for x, y in user.model_dump().items()
+            if x not in 'hash' and y is not None
+        )
+    )
+    computed_hash = hmac.new(
+        settings.telegram_bot_token_hash.digest(),
+        data_check_string.encode(),
+        'sha256',
+    ).hexdigest()
+    is_correct = hmac.compare_digest(computed_hash, user.hash)
+    if not is_correct:
+        raise exps.USER_IS_CORRECT
 
-    auth_token = tkn_manager.create_auth_token({"id": str(user.id)})
+    model_user = models.User(**user.model_dump())
+    if cur_user := await db.user.retrieve(user.id):
+        await db.user.update(cur_user.id, **model_user.model_dump())
+    else:
+        await db.user.create(model_user)
+    await db.session.commit()
+    auth_token = tkn_manager.encode_token({'id': user.id, 'type': 'auth'}, 15)
     return models.AuthToken(auth_token=auth_token)

app/api/endpoints/tokens/pair.py

@@ -1,17 +1,30 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
+from typing_extensions import Annotated
 
 from app import models
-from app.core.security import tkn_manager
+from app.api import deps
+from app.core import exps
+from app.core.security import JWTTokenManager
 
 router = APIRouter()
 
 
-@router.post("/pair/", response_model=models.PairTokens)
-async def new_pair_tokens(data: models.AuthToken):
+@router.post('/pair/', response_model=models.PairTokens)
+async def new_pair_tokens(
+    data: models.AuthToken,
+    tkn_manager: Annotated[JWTTokenManager, Depends(deps.get_tkn_manager)],
+):
     """
-    Получить парные токены
+    Get pair tokens
     """
-    payload = tkn_manager.decode_auth_token(data.auth_token)
-    long_token = tkn_manager.create_long_token(payload)
-    short_token = tkn_manager.create_short_token(payload)
-    return models.PairTokens(long_token=long_token, short_token=short_token)
+
+    payload = tkn_manager.decode_token(data.auth_token)
+    if payload.get('type') != 'auth':
+        raise exps.TOKEN_INVALID
+    payload['type'] = 'access'
+    access_token = tkn_manager.encode_token(payload, 120)
+    payload['type'] = 'refresh'
+    refresh_token = tkn_manager.encode_token(payload, 1440)
+    return models.PairTokens(
+        access_token=access_token, refresh_token=refresh_token
+    )