FastAPI 9. Logout и проверка авторизации

Начиная со статьи "FastAPI 5. Приложение аутентификации и Pydantic схемы", мы с вами начали разработку приложения для аутентификации. И вот, наконец, подходим к завершению этого блока!

В этой статье мы реализуем:

• маршрут для выхода пользователя из системы (logout);
• маршрут для получения текущего авторизованного пользователя;
• зависимость, проверяющую, авторизован ли пользователь.

Теперь давайте разберёмся, как всё это работает.

Зависимость get_current_user

В прошлой статье "FastAPI 8. Маршрут авторизации и JWT", мы познакомились со способами авторизации и выбрали вариант хранения токена в cookies. Теперь, чтобы проверять, авторизован ли пользователь, мы создадим зависимость, которую будем подключать ко всем защищённым маршрутам — тем, где требуется наличие авторизации.

Подробнее про зависимости можно прочитать в статье "FastAPI 5. Приложение аутентификации и Pydantic схемы".

Расшифровка токена

Перейдём в пакет auth и откроем файл handlers.py. Именно здесь у нас расположен класс-обработчик, отвечающий за работу с паролями и токенами.

Добавим в этот класс новый метод — decode_access_token. Он будет отвечать за расшифровку полученного JWT-токена. Метод принимает один аргумент — token (строку) и возвращает словарь.

Как работает метод?

Внутри метода используем конструкцию try-except.

Блок try: 
В теле блока вызываем метод .decode() у импортированного объекта jwt. В него передаём три аргумента:

• jwt — собственно токен, который нужно расшифровать;
• key — секретный ключ, использованный для создания токена. У нас он хранится в поле класса: self.secret;
• algorithms — список допустимых алгоритмов расшифровки. Указываем ["HS256"].

Если токен корректный, метод вернёт словарь с расшифрованными данными, включая user_id и session_id, которые мы в него зашили при генерации.

Блоки except jwt.ExpiredSignatureError и except jwt.InvalidTokenError: 
Если в процессе расшифровки возникнет ошибка, это означает, что токен либо просрочен, либо недействителен. Для обработки таких ситуаций мы используем два отдельных блока except, каждый из которых обрабатывает определённый тип ошибки:

• jwt.ExpiredSignatureError — токен просрочен;
• jwt.InvalidTokenError — токен недействителен или вообще не является JWT.

В обоих случаях мы поднимаем исключение HTTPException, передавая в него статус 401 UNAUTHORIZED и понятное описание ошибки.

Код метода:

import jwt  
from fastapi import HTTPException  
from starlette import status


async def decode_access_token(self, token: str) -> dict:  
    try:  
        return jwt.decode(jwt=token, key=self.secret, algorithms=["HS256"])  
    except jwt.ExpiredSignatureError:  
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has expired"
        )  
    except jwt.InvalidTokenError:  
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )

Этот метод станет основой для проверки авторизации пользователя. В следующих разделах мы на его базе построим зависимость get_current_user.

Получение сессии и пользователя

Теперь нам нужно реализовать два метода — один для работы с Redis, второй для обращения к базе данных. Они понадобятся, чтобы удостовериться, что токен актуален, и получить данные пользователя по user_id.

Откроем файл managers.py в пакете auth.

Метод получения токена из Redis

Даже если в cookies пользователя лежит токен, это ещё не гарантирует, что он действителен. Токен может быть удалён, заблокирован или просрочен. Именно поэтому мы сохраняем все активные токены в Redis — в виде ключей вида user_id:session_id. Таким образом, наличие токена в Redis подтверждает его актуальность.

Реализуем метод get_access_token. Он будет принимать два аргумента:

• user_id — UUID или строку, идентификатор пользователя;
• session_id — строку, идентификатор сессии.

Метод возвращает строку токена или None, если токен не найден.

Внутри метода используем асинхронный контекстный менеджер для подключения к Redis (по аналогии с методом store_access_token, который мы писали ранее). Затем вызываем метод .get() у объекта client, передавая ключ в формате "{user_id}:{session_id}".

Код метода:

async def get_access_token(self, user_id: uuid.UUID | str, session_id: str) -> str | None:  
    async with self.redis.get_client() as client:  
        return await client.get(f"{user_id}:{session_id}")

Этот метод позволит нам проверить, действительно ли сессия существует и активна. Если токен не найден в Redis — считаем сессию недействительной.

Метод получения пользователя по ID

Второй необходимый метод — это получение пользователя из базы данных по его идентификатору.

Он будет очень похож на уже реализованный ранее метод get_user_by_email. Да, можно было бы обобщить их в универсальный метод, но пока не будем усложнять — наша цель сейчас в том, чтобы вы поняли сам принцип работы.

В файле managers.py в пакете auth добавим новый метод get_user_by_id. Он принимает аргумент user_id, который может быть как строкой, так и объектом UUID, и возвращает объект схемы UserVerifySchema. Эту схему мы создадим сразу после метода.

Внутри метода:

1. Открываем асинхронный контекстный менеджер для сессии с базой данных — это уже привычный для нас шаблон.
2. Формируем SQL-запрос с помощью функции select(), указывая нужные поля (на данный момент — только id и email).
3. Добавляем условие с помощью .where(), чтобы выбрать запись по переданному user_id.
4. Выполняем запрос методом session.execute().
5. Извлекаем результат в виде словаря с помощью .mappings().one_or_none() — так мы получим None, если пользователя с таким ID нет.
6. Если пользователь найден, возвращаем объект схемы UserVerifySchema, иначе — None.

Код метода:

async def get_user_by_id(self, user_id: uuid.UUID | str) -> UserVerifySchema | None:  
    async with self.db.db_session() as session:  
        query = select(self.model.id, self.model.email).where(self.model.id  user_id)  

        result = await session.execute(query)  
        user = result.mappings().one_or_none()  

        if user:  
            return UserVerifySchema(**user)  

        return None

Схема UserVerifySchema

Теперь создадим схему, которая будет использоваться для проверки пользователя.

Откроем файл schemas.py в пакете auth и добавим новый класс UserVerifySchema.

Эта схема будет объединять поля из двух других схем — GetUserByID и GetUserByEmail. Мы унаследуемся от них, чтобы избежать дублирования кода.

Дополнительно добавим поле session_id, так как оно будет использоваться для проверки активной сессии. Оно может быть строкой или UUID, а также может отсутствовать, поэтому укажем значение по умолчанию None.

Код класса:

class UserVerifySchema(GetUserByID, GetUserByEmail):  
    session_id: uuid.UUID | str | None = None

Таким образом, у нас теперь есть всё необходимое, чтобы извлечь из базы пользователя и проверить наличие активной сессии в Redis.

Утилита получения токена из Cookies

Чтобы извлекать access-токен из cookie входящего HTTP-запроса, создадим небольшую вспомогательную функцию. Это нужно для того, чтобы не повторять один и тот же код в разных местах, а централизованно обрабатывать логику извлечения токена.

Шаги:

1. Создаём файл utils.py в пакете auth, если он ещё не существует.
2. В этом файле объявим асинхронную функцию get_token_from_cookies.
3. Функция принимает один аргумент — request типа Request из Starlette (FastAPI использует Starlette под капотом).
4. Пытаемся получить токен из cookie по ключу "Authorization".
5. Если токен не найден (значение None), выбрасываем HTTPException со статусом 401 и соответствующим сообщением.
6. Если токен есть — возвращаем его.

Код функции:

from fastapi import HTTPException  
from starlette import status  
from starlette.requests import Request  


async def get_token_from_cookies(request: Request) -> str:  
    token = request.cookies.get("Authorization")  
    if token is None:  
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token is missing")  
    return token

Зависимость для проверки авторизации пользователя

Подготовка завершена — осталось описать саму функцию-зависимость, которая будет проверять авторизацию пользователя.

В пакете auth создадим файл depends.py.

В этом файле опишем асинхронную функцию get_current_user, которая принимает три аргумента:

• token — это строка, содержащая токен авторизации. Однако при вызове функции сам токен ещё находится в cookies. Чтобы получить его оттуда и передать в аргумент, используем специальный тип Annotated[]. Внутрь него первым аргументом передаём ожидаемый тип (str), а вторым — через Depends(...) — вызываем ранее созданную утилиту get_token_from_cookies;
• handler — объект класса AuthHandler. Он создаётся автоматически при помощи Depends, и отвечает за обработку токена;
• manager — объект класса UserManager, также создаётся с помощью Depends и отвечает за работу с пользователями и сессиями.

Функция возвращает объект схемы пользователя — UserVerifySchema.

async def get_current_user(  
    token: Annotated[str, Depends(get_token_from_cookies)],  
    handler: AuthHandler = Depends(AuthHandler),  
    manager: UserManager = Depends(UserManager),  
) -> UserVerifySchema:

Теперь разберём, что происходит внутри этой функции.

Сначала декодируем токен, вызвав метод decode_access_token у объекта handler:

decoded_token = await handler.decode_access_token(token=token)

Из расшифрованного токена извлекаем два значения: user_id и session_id, используя метод .get():

user_id = decoded_token.get("user_id")  
session_id = decoded_token.get("session_id")

Теперь важно проверить, существует ли такая сессия. Для этого вызываем метод get_access_token у manager, передаём в него user_id и session_id, и инвертируем результат при помощи not, чтобы поймать случай, когда токен не найден (или уже удалён):

if not await manager.get_access_token(user_id=user_id, session_id=session_id):  
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token is invalid")

Если токен валиден, нужно получить самого пользователя. Для этого вызываем метод get_user_by_id, передав user_id. Обратите внимание, что user_id приводим к типу UUID:

user = await manager.get_user_by_id(user_id=uuid.UUID(user_id))

Если пользователь не найден (функция вернула None), также поднимаем исключение:

if user is None:  
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")

Если всё хорошо, то перед возвратом добавим к объекту пользователя текущий session_id — он может пригодиться позже (например, при выходе пользователя или продлении сессии):

user.session_id = session_id  

И, наконец, возвращаем пользователя:

return user

Полный код функции:

import uuid  
from typing import Annotated  

from fastapi import Depends, HTTPException  
from starlette import status  

from lkeep.apps.auth.handlers import AuthHandler  
from lkeep.apps.auth.managers import UserManager  
from lkeep.apps.auth.schemas import UserVerifySchema  
from lkeep.apps.auth.utils import get_token_from_cookies  


async def get_current_user(  
    token: Annotated[str, Depends(get_token_from_cookies)],  
    handler: AuthHandler = Depends(AuthHandler),  
    manager: UserManager = Depends(UserManager),  
) -> UserVerifySchema:  
    decoded_token = await handler.decode_access_token(token=token)  
    user_id = decoded_token.get("user_id")  
    session_id = decoded_token.get("session_id")  

    if not await manager.get_access_token(user_id=user_id, session_id=session_id):  
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token is invalid")  

    user = await manager.get_user_by_id(user_id=uuid.UUID(user_id))  
    if user is None:  
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")  

    user.session_id = session_id  

    return user

Маршрут выхода из системы (logout)

Переходим к маршрутам. Первым реализуем маршрут выхода пользователя из системы.

Алгоритм действий:

1. Пользователь на фронтенде нажимает кнопку "Выйти";
2. Фронт отправляет запрос на бэкенд;
3. Запрос поступает на маршрут /logout;
4. Выполняется проверка авторизации и получение текущего пользователя;
5. Сессия удаляется из хранилища, а cookie отзывается;
6. Возвращается ответ, не содержащий данных авторизации.

Функция маршрута

Откроем файл routes.py в пакете auth и добавим новую асинхронную функцию logout, которая принимает два аргумента:

• user — как и ранее, через Annotated получаем текущего пользователя. Указываем схему UserVerifySchema и зависимость get_current_user;
• service — объект класса UserService, получаем через Depends.

Функция возвращает JSONResponse.

Оборачиваем функцию в декоратор @auth_router.get, указывая путь /logout и статус-код 200 OK.

В теле функции вызываем метод logout_user у объекта service, передавая в него пользователя.

Код функции:

@auth_router.get(path="/logout", status_code=status.HTTP_200_OK)  
async def logout(  
    user: Annotated[UserVerifySchema, Depends(get_current_user)],  
    service: UserService = Depends(UserService),  
) -> JSONResponse:  
    return await service.logout_user(user=user)

Метод сервиса

Теперь опишем логику выхода в слое сервиса. Откроем файл services.py в пакете auth.

Добавим метод logout_user, принимающий аргумент user типа UserVerifySchema и возвращающий JSONResponse.

Внутри метода вызываем revoke_access_token у self.manager, передавая идентификаторы пользователя и сессии.

Затем создаём объект JSONResponse с сообщением "Logged out".

После этого вызываем метод .delete_cookie() у объекта ответа и указываем имя удаляемой cookie — Authorization.

В конце возвращаем готовый ответ.

Код метода:

async def logout_user(self, user: UserVerifySchema) -> JSONResponse:  
    await self.manager.revoke_access_token(user_id=user.id, session_id=user.session_id)  

    response = JSONResponse(content={"message": "Logged out"})  
    response.delete_cookie(key="Authorization")  

    return response

Метод менеджера

Теперь реализуем последний метод в цепочке — отзыв токена. Откроем файл managers.py в пакете auth.

Создадим асинхронный метод revoke_access_token, принимающий user_id и session_id. Метод ничего не возвращает.

В теле метода открываем асинхронный контекстный менеджер для работы с Redis. Внутри вызываем метод .delete() у клиента Redis, передав ключ, составленный из идентификаторов пользователя и сессии.

Код метода:

async def revoke_access_token(self, user_id: uuid.UUID | str, session_id: str) -> None:  
    async with self.redis.get_client() as client:  
        await client.delete(f"{user_id}:{session_id}")

Маршрут получения текущего пользователя

Этот маршрут — финальный элемент в нашей системе аутентификации. Он позволяет фронтенду получить информацию о текущем авторизованном пользователе. И, к счастью, это один из самых простых маршрутов.

Открываем файл routes.py в пакете auth и добавляем функцию get_auth_user.

Что делает функция:

• Принимает один аргумент — user, который автоматически передаётся через зависимость get_current_user, используя Annotated и схему UserVerifySchema;
• Возвращает этот объект пользователя;
• Оборачивается в декоратор @auth_router.get, где указывается:
  • путь: /get-user;
  • метод: GET;
  • статус-код: 200 OK;
  • модель ответа: UserVerifySchema — она и будет сериализована и возвращена клиенту.

Код функции:

@auth_router.get(path="/get-user", status_code=status.HTTP_200_OK, response_model=UserVerifySchema)  
async def get_auth_user(user: Annotated[UserVerifySchema, Depends(get_current_user)]) -> UserVerifySchema:  
    return user

Теперь, при запросе на /get-user, если пользователь авторизован, он получит свои данные. Если нет — сработает защита зависимости get_current_user, и будет возвращена ошибка авторизации.

Заключение

Поздравляю — мы завершили реализацию полноценного приложения аутентификации! Конечно, это не финал: мы ещё будем возвращаться к этому модулю, дополнять его новыми возможностями и использовать в других частях проекта. Но уже сейчас у вас есть стабильная и рабочая система входа, регистрации, подтверждения почты, выхода из аккаунта и получения текущего пользователя.

На этом этапе мы научились:

• создавать и использовать зависимости в FastAPI;
• строить маршруты с аутентификацией;
• реализовывать асинхронные сервисы и менеджеры;
• работать с Redis;
• использовать Celery для фоновых задач;
• и, главное, собирать всё это в модульное и масштабируемое приложение.

Всё, что сделали — это каркас, на который можно уверенно опираться при создании любых последующих приложений в FastAPI. Да, со временем запросы к базе станут сложнее, бизнес-логика — объемнее, но базовые принципы останутся прежними.

В следующей части мы перейдём к разработке приложения профиля, в котором дадим пользователю возможность редактировать свои данные.

Репозиторий проекта в GitHub: https://github.com/proDreams/lkeep 
Репозиторий проекта в "GIT на салфетке": https://git.pressanybutton.ru/proDream/lkeep