src/modules/endpoints/endpoints.py - master - sam_single_1/ServiceVkAuth

src/modules/endpoints/endpoints.py
214 строк6 KB
sam_single_1
remove errors
23 окт 2024, 17:34
23 окт 2024, 17:345aebf78
from fastapi import APIRouter, Response, Body, Cookie, Header
from pydantic import BaseModel, Field
from typing import Annotated, Dict

from dto.endpoints_adapter_to_endpoints import (
    CommonOutput,
    FirstLoginOutput,
    SecondLoginInput,
    SecondLoginOutput,
    LogoutInput,
    LogoutOutput,
    ValidateInput,
    ValidateOutput,
    RefreshInput,
    RefreshOutput,
)

from src.logic.endpoints_handling import (
    logic_first_login,
    logic_login_second,
    logic_logout,
    logic_refresh,
    logic_validate,
)

from modules.endpoints.error_mapper import error_mapper
from modules.endpoints.error_shecmas import (
    ErrorResponse400,
    ErrorResponse401,
    ErrorResponse403,
)

auth_router = APIRouter(prefix="", tags=["auth"])
validate_router = APIRouter(prefix="", tags=["validation"])


class FirstLoginResponse(BaseModel):
    code_challenge: Annotated[str, Field()]
    state: Annotated[str, Field()]


@auth_router.get(
    "/login/first",
    response_model=FirstLoginResponse,
    description="""
## Для начала регистрации обратитесь к этому эндпоинту 
Тогда этот эндпоинт:
1. Иницилизирует процесс авторизации   
2. Генерирует набор параметров PKCE
3. Возвращает state и code_challenge
""",
    summary="Инициализация процесс авторизации",
    responses={"400": {"model": ErrorResponse400}},
)
async def first_step(response: Response):
    output: CommonOutput[FirstLoginOutput] = await logic_first_login()
    if output.has_error:
        return error_mapper(output, res=response)
    return FirstLoginResponse(
        code_challenge=output.OkOutput.code_challenge, state=output.OkOutput.state
    )


class SecondLoginRequest(BaseModel):
    code: Annotated[str, Field()]
    state: Annotated[str, Field()]
    device_id: Annotated[str, Field()]


class SecondLoginResponse(BaseModel):
    access_token: Annotated[str, Field()]
    user_id: Annotated[int, Field()]


@auth_router.post(
    "/login/second",
    tags=["auth"],
    response_model=SecondLoginResponse,
    responses={"400": {"model": ErrorResponse400}},
    description="""
## Для завершения регистрации обратитесь к этому эндпоинту 
Тогда этот эндпоинт:
1. По state понимает кто вы 
2. Меняет ваш авторизационный код на токены авторизации
3. Возвращает access_token и ставит httpOnly Cookie для refresh_token
""",
    summary="Завершает авторизации",
)
async def second_step(
    date: Annotated[SecondLoginRequest, Body()],
    response: Response,
):
    input = SecondLoginInput(code=date.code, state=date.state, device_id=date.state)
    output: CommonOutput[SecondLoginOutput] = await logic_login_second(input)
    if output.has_error:
        return error_mapper(output, res=response)
    output_ok = output.OkOutput
    resp = SecondLoginResponse(
        access_token=output_ok.access_token, user_id=output_ok.user_id
    )
    response.set_cookie(
        key="Authorization",
        value=output_ok.refresh_token,
        httponly=True,
        expires=output_ok.expires_access_token_at,
        secure=True,
    )

    return resp


class AuthHeaderLogout(BaseModel):
    AccessToken: str


class LogoutResponse(BaseModel):
    pass


@auth_router.get(
    "/logout",
    tags=["auth"],
    responses={"400": {"model": ErrorResponse400}, "401": {"model": ErrorResponse401}},
    description="""
## Для выхода из аккаунта обратитесь к этому эндпоинту 
Тогда этот эндпоинт:
1. Проверит авторизованы ли вы  
2. Деавторизует вас из VK ID
3. Удалит ваш refresh_token
""",
    summary="Выход из аккаунта",
)
async def logout(
    access_token: Annotated[AuthHeaderLogout, Header()],
    response: Response,
):
    input = LogoutInput(access_token=access_token.AccessToken)
    output: CommonOutput[LogoutOutput] = await logic_logout(input)
    if output.has_error:
        return error_mapper(output, res=response)
    response.delete_cookie(key="refresh_token")
    return LogoutResponse()


class AuthCookie(BaseModel):
    Authorization: str


class AuthHeader(BaseModel):
    AccessToken: str


class RefreshResponse(BaseModel):
    pass


@auth_router.get(
    "/refresh",
    tags=["auth"],
    response_model=RefreshResponse,
    responses={"400": {"model": ErrorResponse400}, "401": {"model": ErrorResponse401}},
    description="""
## По истечение срока жизни access_token обратитесь к этому эндпоинты
Тогда этот эндпоинт:
1. Проверит авторизованы ли вы  
2. Обратится к VK ID для нового токена
3. Вернет вам обновеленные токены
""",
    summary="Обновление access _token",
)
async def refresh(
    response: Response,
    token: Annotated[AuthCookie, Cookie()],
):
    input: RefreshInput = RefreshInput(refresh_token=token.Authorization)
    output: CommonOutput[RefreshOutput] = await logic_refresh(i=input)
    if output.has_error:
        return error_mapper(output, res=response)
    return RefreshResponse()


class AuthHeaderValidate(BaseModel):
    AccessToken: str


class ValidateResponse(BaseModel):
    pass


@validate_router.get(
    "/validate",
    tags=["validation"],
    responses={
        "400": {"model": ErrorResponse400},
        "401": {"model": ErrorResponse401},
        "403": {"model": ErrorResponse403},
    },
    description="""
## Для выхода из аккаунта обратитесь к этому эндпоинту 
Тогда этот эндпоинт:
1. Получает ваш токен авторизации в заголове  
2. Проверяет валидность вашего токена
3. Возвращает user_id, если вы авторизованы
""",
    summary="Проверит авторизованы ли вы",
)
async def validate(response: Response, access_token: Annotated[AuthHeader, Header()]):
    input = ValidateInput(access_token=access_token.AccessToken)
    ouput: CommonOutput[ValidateOutput] = await logic_validate(i=input)
    if ouput.has_error:
        return error_mapper(ouput, res=response)
    else:
        response.headers["User_id"] = ouput.OkOutput.user_id
        return ValidateResponse()