FastAPI 6. Пользовательский сервис и маршруты регистрации

Мы подошли к одной из ключевых тем — маршрутам. Чтобы обращаться к API и выполнять нужные действия, необходимо настроить маршруты. Но сами маршруты, по сути, лишь принимают запросы. Вся бизнес-логика должна быть вынесена в специализированный компонент — пользовательский сервис.

Что такое пользовательский сервис?

Пользовательский сервис можно сравнить с менеджером, но у них разные роли:

• Менеджер отвечает за взаимодействие с базой данных.
• Сервис выступает промежуточным звеном между маршрутом и менеджером.

Основная задача сервиса — содержать всю логику, которая выполняется при вызове маршрута. Таким образом, маршрут просто получает запрос, передаёт данные в сервис и возвращает результат. Этот подход позволяет сделать код более структурированным и тестируемым. Хотя иногда логика может быть немного другой, но такой стиль — это хороший базис.

Создание пользовательского сервиса

Для начала создадим файл services.py в пакете auth. Класс UserService будет отвечать за реализацию функционала, связанного с пользователями.

Конструктор класса

Начнём с определения конструктора. Создадим метод __init__, который принимает два аргумента:

• manager — объект класса UserManager, отвечающего за работу с базой данных.
• handler — объект класса AuthHandler, выполняющего задачи, связанные с авторизацией.

Оба аргумента будут передаваться через зависимости FastAPI с помощью Depends. Это упрощает процесс инъекции зависимостей и делает сервис более гибким.

Вот как выглядит конструктор класса:

def __init__(  
    self,  
    manager: UserManager = Depends(UserManager),  
    auth: AuthHandler = Depends(AuthHandler),  
) -> None:  
    self.manager = manager  
    self.auth = auth

Метод register_user

Теперь добавим в сервис метод для регистрации нового пользователя.

Этот метод:

• Будет асинхронным.
• Примет объект user класса схемы RegisterUser.
• Вернёт объект класса схемы UserReturnData.

Логика метода:

1. Получить хэш пароля пользователя с помощью метода .get_password_hash из объекта handler.
2. Создать объект пользователя new_user с хэшированным паролем и email.
3. Передать этого пользователя в метод .create_user менеджера для сохранения в базе данных.
4. Вернуть результат.

Пример реализации:

async def register_user(self, user: RegisterUser) -> UserReturnData:  
    # Хэшируем пароль
    hashed_password = await self.handler.get_password_hash(user.password)  

    # Создаём объект пользователя
    new_user = CreateUser(email=user.email, hashed_password=hashed_password)  

    # Сохраняем пользователя в базе и возвращаем результат
    return await self.manager.create_user(user=new_user)

Полный код файла:

from fastapi import Depends  

from lkeep.apps.auth.handlers import AuthHandler  
from lkeep.apps.auth.managers import UserManager  
from lkeep.apps.auth.schemas import RegisterUser, UserReturnData, CreateUser  


class UserService:  
    def __init__(self, manager: UserManager = Depends(UserManager), handler: AuthHandler = Depends(AuthHandler)):  
        self.manager = manager  
        self.handler = handler  

    async def register_user(self, user: RegisterUser) -> UserReturnData:  
        hashed_password = await self.handler.get_password_hash(user.password)  

        new_user = CreateUser(email=user.email, hashed_password=hashed_password)  

        return await self.manager.create_user(user=new_user)

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

Маршрут регистрации

Маршруты, они же эндпоинты (endpoints) или "ручки", — это пути, по которым клиент может обращаться к вашему API. Например, site.ru/api/v1/get_users. Каждый маршрут также определяет метод HTTP-запроса: GET, POST, PUT, DELETE и другие. Если вы не знакомы с HTTP-методами, настоятельно рекомендую прочитать об этом перед тем, как двигаться дальше, например, в статье на Хабр.

Создание маршрута регистрации

Шаг 1. Создаём роутер

В пакете auth создадим файл routes.py. Этот файл будет содержать маршруты, относящиеся только к приложению аутентификации.

Внутри файла создадим переменную auth_router, которая будет экземпляром класса APIRouter. Это специальный инструмент FastAPI, предназначенный для группировки маршрутов.

Укажем следующие параметры при создании роутера:

• prefix: общий префикс для всех маршрутов в роутере (например, /auth).
• tags: список тегов, помогающий группировать маршруты в документации Swagger.

Пример кода:

from fastapi import APIRouter

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

Важно: при указании префикса (prefix) обязательно добавляйте / в начале строки! Это требование FastAPI.

Шаг 2. Создаём функцию-обработчик для маршрута

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

• user: объект схемы RegisterUser, представляющий данные, которые отправляет клиент.
• service: экземпляр UserService, передаваемый через зависимость (Depends).

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

Пример кода:

from fastapi import Depends
from lkeep.apps.auth.schemas import RegisterUser, UserReturnData
from lkeep.apps.auth.services import UserService

async def registration(user: RegisterUser, service: UserService = Depends(UserService)) -> UserReturnData:
    return await service.register_user(user=user)

Шаг 3. Добавляем маршрут в роутер

Чтобы функция-обработчик стала полноценным маршрутом, нужно использовать декоратор, добавляющий её в роутер. Для этого обращаемся к методу .post объекта auth_router.

Передаём в декоратор следующие параметры:

• path: путь маршрута относительно префикса (в данном случае "/register").
• response_model: схема данных, возвращаемая в ответе.
• status_code: HTTP-статус ответа. Здесь указан 201 CREATED, который используется для подтверждения успешного создания ресурса.

Итоговый код функции с декоратором:

from starlette import status

@auth_router.post(
    "/register",
    response_model=UserReturnData,
    status_code=status.HTTP_201_CREATED
)
async def registration(
    user: RegisterUser, 
    service: UserService = Depends(UserService)
) -> UserReturnData:
    return await service.register_user(user=user)

Полный код файла routes.py

Вот как выглядит полный файл с маршрутом регистрации:

from fastapi import APIRouter, Depends  
from starlette import status  

from lkeep.apps.auth.schemas import RegisterUser, UserReturnData  
from lkeep.apps.auth.services import UserService  

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

@auth_router.post(
    "/register", 
    response_model=UserReturnData, 
    status_code=status.HTTP_201_CREATED
)  
async def registration(
    user: RegisterUser, 
    service: UserService = Depends(UserService)
) -> UserReturnData:  
    return await service.register_user(user=user)

Что делает этот код?

1. Роутер auth_router: отвечает за группировку всех маршрутов, относящихся к аутентификации.
2. Маршрут POST /auth/register:
   • Получает данные пользователя.
   • Передаёт их в UserService для обработки.
   • Возвращает результат в виде объекта UserReturnData.
3. Swagger-документация: благодаря tags и response_model, маршрут автоматически отображается в Swagger с понятным описанием.

Роутер приложений

На предыдущем шаге мы создали роутер auth_router для маршрутов, связанных с аутентификацией. Теперь нам нужен общий роутер, который объединяет все маршруты разных приложений в проекте.

Создаём общий роутер

Зачем нужен общий роутер?

Общий роутер позволяет централизованно управлять всеми маршрутами вашего проекта. Например:

• Устанавливать общий префикс (/api/v1), чтобы все маршруты проекта начинались с него.
• Упрощать интеграцию новых модулей (например, orders, products и т.д.).
• Обеспечивать совместимость разных версий API (например, /api/v1 для старой версии и /api/v2 для новой).

Это особенно важно в крупных или развивающихся проектах, где маршруты могут меняться или расширяться.

Добавляем общий роутер

В файле __init__.py пакета apps создадим общий роутер apps_router. Этот роутер будет включать в себя роутеры всех приложений.

Код:

from fastapi import APIRouter

apps_router = APIRouter(prefix="/api/v1")

Почему v1? 
Использование версии API (v1) не обязательно, но это хороший стандарт. Если в будущем вы внесёте серьёзные изменения в API, вы сможете создать новую версию (/api/v2), сохранив старую для обратной совместимости. Это особенно важно для публичных API.

Подключаем роутер приложения auth

Чтобы включить маршруты auth_router в общий роутер, используем метод .include_router. Вызов выглядит так:

apps_router.include_router(router=auth_router)

Здесь:

• router — это аргумент, который указывает на роутер, который мы подключаем (в данном случае, auth_router).
• Теперь все маршруты из auth_router будут доступны в рамках apps_router.

Полный код файла __init__.py

Файл __init__.py пакета apps будет выглядеть так:

from fastapi import APIRouter  

from lkeep.apps.auth.routes import auth_router  

apps_router = APIRouter(prefix="/api/v1")  

# Подключаем маршруты приложения auth
apps_router.include_router(router=auth_router)

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

1. Создаётся роутер apps_router: с общим префиксом /api/v1.
2. Подключаются маршруты приложения auth: через метод .include_router.
3. Объединение маршрутов: Теперь маршруты приложения auth будут доступны с полным путём /api/v1/auth, где:
   • /api/v1 — префикс общего роутера.
   • /auth — префикс роутера приложения auth.

Почему это важно?

• Удобство: Все маршруты приложений централизованно собираются в одном месте.
• Гибкость: Легко добавить новые модули, просто подключив их роутеры.
• Масштабируемость: Если проект вырастет, вы сможете поддерживать несколько версий API без лишних проблем.

Главный роутер

Последний шаг — подключить роутер из пакета apps в главный файл проекта.

Вносим изменения в main.py

Откроем файл main.py для редактирования.

В файле main.py уже есть переменная app, которая является экземпляром класса FastAPI. Чтобы зарегистрировать роутеры, используем метод include_router:

app.include_router(router=apps_router)

Здесь:

• router — аргумент, указывающий на объект роутера (apps_router), который объединяет маршруты всех приложений.

Полный код файла main.py

import uvicorn  
from fastapi import FastAPI  

from lkeep.apps import apps_router  

app = FastAPI()  

app.include_router(router=apps_router)  

def start():  
    uvicorn.run(app="lkeep.main:app", reload=True)

Теперь маршруты из всех подключённых приложений будут доступны через ваш API.

Запуск!

Чтобы запустить приложение, выполните команду:

poetry run app

После запуска веб-сервера откройте Swagger по адресу: http://127.0.0.1:8000/docs.

Тестирование маршрута регистрации

Перейдите к маршруту /auth/register в Swagger-документации.

Нажмите кнопку "Try it out".

Введите данные пользователя:

• email — например, test@example.com.
• password — любой желаемый пароль.

Нажмите "Execute".

После успешного выполнения вы увидите:

• Ответ сервера: информацию о созданном пользователе.
• База данных: новая запись с введённым email и хешированным паролем.

Заключение

На этом базовая настройка приложения завершена. Однако это только начало пути. Впереди нас ждут новые вызовы:

• Реализация авторизации через токены.
• Настройка сброса пароля.
• Подтверждение регистрации через email.
• Основной функционал приложения.

Это сложный, но невероятно увлекательный процесс, который поможет вам вырасти как разработчику.