FastApi: авторизированный вход

Из предыдущей статьи у нас есть некий вызов API после выполнения которого у нас на руках будет некий сессионый ключ.

Попробуем теперь используя этот сессионый ключ, запросить некую информацию о пользователе. Для этого добавим в схему auth.py классы:

class info_class(BaseModel):
    username : Optional[str] = Field(None, title="Имя пользователя", example="Вася Пупкин")
    phone    : Optional[str] = Field(None, title="Номер телефона", example="+79222347594")
    email    : Optional[str] = Field(None, title="Email пользователя", example="dowiurefh@mail.ru")

class info_out(BaseModel):
    error     : Optional[bool]  = Field(..., title="Результат выполнения запроса", example="false")
    comment   : Optional[str]   = Field(None, title="Текст ошибки", example="Пользователь удалён")
    result    : Optional[info_class]

Следующим шагом будет защитить доступ к остальным компонентам API, от доступа без сессионного ключа. Для этого воспользуемся классом FastApi security и добавим в требования заголовка запроса тег авторизации.

Получим итоговый роутинг auth.py:

# -*- coding: utf-8 -*-
"""Different helper-functions to work with users."""
import json
import fastapi;
from fastapi import APIRouter,Depends
from app.schemas import auth as auth_models
from pydantic import BaseModel

token_key = fastapi.security.APIKeyHeader(name="Authorization")

class Token(BaseModel):
    token: str

# получить текущий токен
def get_current_token(oauth_header: str = fastapi.Security(token_key)):
    #print(f"Token: {oauth_header}")
    oauth_header=oauth_header.replace("Token ","")
    return oauth_header

router = APIRouter()

@router.post('/auth',response_model=auth_models.auth_out,tags=["auth"]) # заданы исходящие параметры
async def auth(into = Depends(auth_models.auth_in)): # заданы входящие параметры
    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash="12345678"
    # сформируем ответ
    out=auth_models.auth_out(hash=hash)
    return out

@router.post('/GetInfo',response_model=auth_models.info_out,tags=["auth"]) # заданы исходящие параметры
async def GetInfo(token: Token=Depends(get_current_token)): # заданы входящие параметры
    print(f"Токен: {token}")
    if token!="12345678":
        print("Токен не верен!")
        return auth_models.info_out(error=True,comment="Токен не найден")
    # сформируем ответ

    return auth_models.info_out(error=False,result={"username":"Pavel","email":"vasya@mail.ru"})

В результате мы можем увидеть в документации появившийся «замочек»:

При попытке выполнения запроса без авторизации, соответственно получим ошибку аторизации:

Not authenticated

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *

Этот сайт использует Akismet для борьбы со спамом. Узнайте, как обрабатываются ваши данные комментариев.