Refactor token creation logic and fix datetime usage
Adjusted `datetime.utcnow` to `datetime.now` for consistency and refactored token creation functions for cleaner structure. Removed duplicated `create_access_token` and `create_refresh_token` definitions by consolidating them into a single location.
This commit is contained in:
@@ -50,7 +50,6 @@ def create_tokens(user_id: str) -> TokenResponse:
|
|||||||
user_id=user_id
|
user_id=user_id
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def create_token(
|
def create_token(
|
||||||
data: dict,
|
data: dict,
|
||||||
expires_delta: Optional[timedelta] = None,
|
expires_delta: Optional[timedelta] = None,
|
||||||
@@ -60,9 +59,9 @@ def create_token(
|
|||||||
to_encode = data.copy()
|
to_encode = data.copy()
|
||||||
|
|
||||||
if expires_delta:
|
if expires_delta:
|
||||||
expire = datetime.utcnow() + expires_delta
|
expire = datetime.now() + expires_delta
|
||||||
else:
|
else:
|
||||||
expire = datetime.utcnow() + (
|
expire = datetime.now() + (
|
||||||
timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) if token_type == "access"
|
timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) if token_type == "access"
|
||||||
else timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
|
else timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
|
||||||
)
|
)
|
||||||
@@ -70,12 +69,20 @@ def create_token(
|
|||||||
to_encode.update({
|
to_encode.update({
|
||||||
"exp": expire,
|
"exp": expire,
|
||||||
"type": token_type,
|
"type": token_type,
|
||||||
"iat": datetime.utcnow(),
|
"iat": datetime.now(),
|
||||||
"jti": str(uuid4())
|
"jti": str(uuid4())
|
||||||
})
|
})
|
||||||
|
|
||||||
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||||
|
|
||||||
|
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
||||||
|
"""Create a new access token."""
|
||||||
|
return create_token(data, expires_delta, "access")
|
||||||
|
|
||||||
|
|
||||||
|
def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
||||||
|
"""Create a new refresh token."""
|
||||||
|
return create_token(data, expires_delta, "refresh")
|
||||||
|
|
||||||
def decode_token(token: str, required_type: str = "access") -> TokenPayload:
|
def decode_token(token: str, required_type: str = "access") -> TokenPayload:
|
||||||
"""
|
"""
|
||||||
@@ -130,15 +137,3 @@ def decode_token(token: str, required_type: str = "access") -> TokenPayload:
|
|||||||
except JOSEError as e: # All other JOSE-related errors
|
except JOSEError as e: # All other JOSE-related errors
|
||||||
raise JWTError("Invalid token.") from e
|
raise JWTError("Invalid token.") from e
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
|
||||||
"""Create a new access token."""
|
|
||||||
return create_token(data, expires_delta, "access")
|
|
||||||
|
|
||||||
|
|
||||||
def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
|
||||||
"""Create a new refresh token."""
|
|
||||||
return create_token(data, expires_delta, "refresh")
|
|
||||||
|
|||||||
Reference in New Issue
Block a user