jwt refresh token 1.0

server/backend/JWT.py

@@ -3,21 +3,47 @@ from jose import JWTError, jwt
 from fastapi import HTTPException, Depends, status
 from fastapi.security import OAuth2PasswordBearer
 
+from server.database import db
+
 from dotenv import load_dotenv                              #Работа с env для jwt
 import os
 load_dotenv()
 SECRET_KEY = os.getenv('SECRET_KEY')
 ALGORITHM = os.getenv('ALGORITHM')
 ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv('ACCESS_TOKEN_EXPIRE_MINUTES'))
-
+REFRESH_TOKEN_EXPIRE_MINUTES = int(os.getenv('REFRESH_TOKEN_EXPIRE_MINUTES'))
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")      #Создание jwt
-async def create_access_token(data: dict, expires_delta: timedelta | None = None):
-    to_encode = data.copy()
-    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
+class Token():
+    @staticmethod
+    async def create_token(data: dict, expires_delta: timedelta | None = None):
+        to_encode = data.copy()
+        expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
+        to_encode.update({"exp": expire})
+        encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+        return encoded_jwt
+class AccessToken(Token):
+    @staticmethod
+    async def create(data:dict, expires_delta: timedelta | None = None):
+        return await Token.create_token(data, expires_delta)
+class RefreshToken(Token):
+    @staticmethod
+    async def create(data:dict, expires_delta: timedelta | None = None):
+        token_str = await Token.create_token(data, expires_delta)
+        await db.refresh_token(encoded_jwt = token_str,email=data["sub"])
+        return token_str
+# async def create_access_token(data: dict, expires_delta: timedelta | None = None):
+#     to_encode = data.copy()
+#     expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
+#     to_encode.update({"exp": expire})
+#     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+#     return encoded_jwt
+# async def create_refresh_token(data:dict, expires_delta:timedelta | None = None):
+#     to_encode = data.copy()
+#     expire = datetime.utcnow() + (expires_delta or timedelta(minutes=6000))
+#     to_encode.update({"exp": expire})
+#     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+#     return encoded_jwt
 
 async def current_user(token: str = Depends(oauth2_scheme)): #Проверка jwt
     try: