qzl
51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Protocol
|
|
|
|
from v1.auth.schemas import (
|
|
AuthTokenResponse,
|
|
AuthUserByEmailResponse,
|
|
LoginRequest,
|
|
RefreshRequest,
|
|
SignupRequest,
|
|
)
|
|
|
|
|
|
class AuthServiceGateway(Protocol):
|
|
async def signup(self, request: SignupRequest) -> AuthTokenResponse:
|
|
raise NotImplementedError
|
|
|
|
async def login(self, request: LoginRequest) -> AuthTokenResponse:
|
|
raise NotImplementedError
|
|
|
|
async def refresh(self, request: RefreshRequest) -> AuthTokenResponse:
|
|
raise NotImplementedError
|
|
|
|
async def logout(self, refresh_token: str | None) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def get_user_by_email(self, email: str) -> AuthUserByEmailResponse:
|
|
raise NotImplementedError
|
|
|
|
|
|
class AuthService:
|
|
_gateway: AuthServiceGateway
|
|
|
|
def __init__(self, gateway: AuthServiceGateway) -> None:
|
|
self._gateway = gateway
|
|
|
|
async def signup(self, request: SignupRequest) -> AuthTokenResponse:
|
|
return await self._gateway.signup(request)
|
|
|
|
async def login(self, request: LoginRequest) -> AuthTokenResponse:
|
|
return await self._gateway.login(request)
|
|
|
|
async def refresh(self, request: RefreshRequest) -> AuthTokenResponse:
|
|
return await self._gateway.refresh(request)
|
|
|
|
async def logout(self, refresh_token: str | None) -> None:
|
|
await self._gateway.logout(refresh_token)
|
|
|
|
async def get_user_by_email(self, email: str) -> AuthUserByEmailResponse:
|
|
return await self._gateway.get_user_by_email(email)
|