FastApi: авторизированный вход
Из предыдущей статьи у нас есть некий вызов API после выполнения которого у нас на руках будет некий сессионый ключ.
Попробуем теперь используя этот сессионый ключ, запросить некую информацию о пользователе. Для этого добавим в схему auth.py классы:
|
1 2 3 4 5 6 7 8 9 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
# -*- 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"}) |
В результате мы можем увидеть в документации появившийся «замочек»:
При попытке выполнения запроса без авторизации, соответственно получим ошибку аторизации:
|
1 |
Not authenticated |