59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
from typing import Optional
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.core.security import create_access_token, create_refresh_token, verify_token
|
|
from src.domain.schemas import Token
|
|
from src.repository.user_repository import UserRepository
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
self.user_repo = UserRepository(db)
|
|
|
|
async def authenticate_user(self, username: str, password: str) -> Optional[Token]:
|
|
user = await self.user_repo.authenticate(username, password)
|
|
if not user or not user.is_active:
|
|
return None
|
|
access_token = create_access_token(data={"sub": user.username})
|
|
refresh_token = create_refresh_token(data={"sub": user.username})
|
|
return Token(
|
|
access_token=access_token,
|
|
refresh_token=refresh_token,
|
|
token_type="bearer",
|
|
)
|
|
|
|
async def authenticate_user_via_email(self, email: str) -> Optional[Token]:
|
|
user = await self.user_repo.authenticate_via_email(email)
|
|
if not user or not user.is_active:
|
|
return None
|
|
access_token = create_access_token(data={"sub": user.email})
|
|
refresh_token = create_refresh_token(data={"sub": user.email})
|
|
return Token(
|
|
access_token=access_token,
|
|
refresh_token=refresh_token,
|
|
token_type="bearer",
|
|
)
|
|
|
|
async def refresh_token(self, refresh_token: str) -> Optional[Token]:
|
|
payload = verify_token(refresh_token)
|
|
if payload is None:
|
|
return None
|
|
|
|
username: str | None = payload.get("sub")
|
|
if username is None:
|
|
return None
|
|
|
|
user = await self.user_repo.get_by_username(username)
|
|
if not user:
|
|
return None
|
|
|
|
access_token = create_access_token(data={"sub": user.username})
|
|
new_refresh_token = create_refresh_token(data={"sub": user.username})
|
|
return Token(
|
|
access_token=access_token,
|
|
refresh_token=new_refresh_token,
|
|
token_type="bearer",
|
|
)
|