FastAPI 10. Изменение данных пользователя

В предыдущих девяти статьях мы 

• спроектировали архитектуру проекта;
• подключили PostgreSQL, Redis и Celery;
• реализовали приложение аутентификации.

В этом материале мы добавим функциональность изменения данных пользователя — его электронной почты и пароля. Я буду называть этот модуль «профилем», однако отдельную модель Profile создавать не станем — по крайней мере на этапе MVP.

В контексте наших статей термин «приложение» обозначает логически обособленный пакет внутри проекта. У него есть собственный router, схемы и сервисы — примерно так же, как это принято в Django. Такой подход помогает держать код компактным и читаемым.

План действий

1. Создать новое приложение profile.
2. Описать маршруты для изменения e‑mail и пароля.
3. Реализовать сервисный слой и менеджер для работы с БД.

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

Поехали!

Новое приложение profile

В каталоге apps создаём пакет profile.

Внутри сразу заводим четыре файла:

Зачем четыре отдельных файла? Такой расклад помогает избежать «гигантов‑модулей», где маршруты, бизнес‑логика и доступ к БД перемешаны. Чёткое разделение повышает читаемость и упрощает тестирование.

Почему отдельное приложение?

Логичный вопрос: зачем выделять «профиль» в отдельный модуль, раз мы всё равно меняем поля пользователя?

Короткий ответ — принцип единой ответственности.

• auth — регистрация, авторизация и выдача JWT‑токенов.
• profile — изменение существующих данных (e‑mail, пароль).

Такое деление упрощает навигацию по проекту и даёт возможность развивать подсистемы независимо.

Менеджер профиля — обновление данных в БД

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

Зачем делать это заранее? 
Если написать отдельные методы смены e-mail и пароля, код получится почти одинаковым: различаться будут лишь изменяемые поля. Дублирование плохо, а потому всё, что можно обобщить, следует обобщать. Мы реализуем единый метод, способный обновлять одно или несколько полей пользователя.

Открываем файл managers.py и добавляем класс ProfileManager.

__init__

Сначала опишем dunder-метод __init__ (его ещё называют конструктором класса), принимающий (кроме self) объект db типа DBDependency, пробрасываемый через Depends.

Внутри сохраняем две переменные:

• self.db — пришедший объект DBDependency;
• self.user_model — ссылка на класс User.

from fastapi import Depends

from lkeep.core.core_dependency.db_dependency import DBDependency
from lkeep.database.models import User


class ProfileManager:
    def __init__(
        self,
        db: DBDependency = Depends(DBDependency),
    ) -> None:
        self.db = db
        self.user_model = User

update_user_fields

Вторым идёт асинхронный метод update_user_fields, которому нужны два аргумента:

• user_id: UUID пользователя;
• **kwargs: произвольные пары «поле → значение» для обновления.

Алгоритм работы:

Пошаговый алгоритм:

1. Открываем контекстный менеджер async with self.db.db_session() as session и получаем AsyncSession.
2. Через update() формируем запрос:
   • указываем модель self.user_model;
   • задаём условие .where(self.user_model.id user_id);
   • передаём новые значения распаковкой **kwargs в .values(**kwargs).
3. Выполняем запрос await session.execute(query).
4. Подтверждаем изменения await session.commit().

import uuid

from sqlalchemy import update


async def update_user_fields(self, user_id: uuid.UUID | str, **kwargs: Any) -> None:  
    async with self.db.db_session() as session:  
        query = update(self.user_model).where(self.user_model.id == user_id).values(**kwargs)  

        await session.execute(query)  

        await session.commit()

Важно: без вызова commit() изменения не сохранятся, и, например, новый e-mail не появится в базе.

Полный код файла вы можете посмотреть в репозитории.

Теперь у нас есть менеджер с универсальным методом, который пригодится и для смены почты, и для смены пароля.

Смена электронной почты

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

Начинать будем не с маршрута, а «с середины» — со схем и сервиса. Так к моменту, когда доберёмся до роутера, у нас уже будут готовы импорты и не придётся возвращаться, чтобы что-то поправить.

Схемы профиля — схема новой почты

Чтобы получать новую электронную почту от пользователя, нам нужна соответствующая Pydantic-схема.

Открываем ранее созданный файл schemas.py.

В нём объявляем единственный класс-модель ChangeEmailRequest, унаследованный от BaseModel. 
Внутри достаточно одного поля:

• new_email — новый адрес пользователя, тип EmailStr (валидирует, что строка похожа на e-mail).

Этой схемы достаточно для операции смены почты: серверу нужен только корректный e-mail, а идентификатор пользователя мы берём из авторизационных данных.

from pydantic import BaseModel, EmailStr


class ChangeEmailRequest(BaseModel):
    new_email: EmailStr

Почему EmailStr? Тип EmailStr из Pydantic сразу проверяет формат адреса и выдаёт ошибку 422, если клиент прислал что-то вроде not-an-email. Это снимает лишнюю валидацию с наших ручных проверок и держит код чище.

Больше схем для смены e-mail нам не требуется.

Сервис профиля — бизнес-логика смены e-mail

Переходим к service-слою. Напомню, сервис отвечает за бизнес-правила и оркестрацию: проверяет входные данные (если нужно), вызывает менеджеры, а роутеру возвращает результат.

Открываем services.py и объявляем ProfileService.

__init__

Как и в ProfileManager, первым идёт dunder-метод __init__, который принимает через Depends экземпляр ProfileManager и сохраняет его в self.manager.

from fastapi import Depends

from lkeep.apps.profile.managers import ProfileManager


class ProfileService:
    def __init__(self, manager: ProfileManager = Depends(ProfileManager)) -> None:
        self.manager = manager

change_email

Далее пишем асинхронный метод change_email. Ему нужны два параметра:

1. data: объект схемы ChangeEmailRequest (новый адрес уже прошёл валидацию как EmailStr);
2. user: объект авторизованного пользователя, например UserVerifySchema, который вы получаете из зависимости get_current_user().

Дополнительных проверок здесь не требуется, поэтому метод просто делегирует работу менеджеру:

from lkeep.apps.profile.schemas import ChangeEmailRequest
from lkeep.apps.auth.schemas import UserVerifySchema


async def change_email(
    self,
    data: ChangeEmailRequest,
    user: UserVerifySchema,
) -> None:
    return await self.manager.update_user_fields(
        user_id=user.id,
        email=data.new_email,
    )

Что делает сервис? Он не «знает» о внутреннем устройстве базы данных и не формирует SQL-запросы. Его задача — принять валидированные данные, вызвать подходящий метод менеджера и при необходимости обработать бизнес-исключения (например, если e-mail уже занят). В нашем MVP эти проверки опущены, но сервисный слой позволит легко добавить их позже.

На этом бизнес-логика смены e-mail закончена. Следующим шагом будет написание маршрута, который свяжет всё вместе.

Маршруты профиля — маршрут смены почты

Теперь опишем HTTP‑маршрут, который будет принимать запросы на смену e‑mail.

Файл routes.py начинается с объявления экземпляра APIRouter. Мы уже делали это в статье «FastAPI 6. Пользовательский сервис и маршруты регистрации», но для полноты повторим здесь:

from fastapi import APIRouter

profile_router = APIRouter(prefix="/profile", tags=["profile"])

В объекте profile_router мы задали:

• prefix="/profile" — все пути модуля будут начинаться с этого префикса;
• tags=["profile"] — тег для автоматической документации Swagger‑UI.
•  

Объявляем эндпоинт /change-email

Следующим шагом добавляем функцию‑обработчик и «вешаем» её на POST‑маршрут /change-email. При успешном обновлении адреса сервер вернёт статус‑код 200 OK (по сути — пустой ответ без тела).

from typing import Annotated

from fastapi import Depends
from starlette import status

from lkeep.apps.auth.depends import get_current_user
from lkeep.apps.auth.schemas import UserVerifySchema
from lkeep.apps.profile.schemas import ChangeEmailRequest
from lkeep.apps.profile.services import ProfileService


@profile_router.post(
    "/change-email",
    status_code=status.HTTP_200_OK,
)
async def change_email(
    data: ChangeEmailRequest,
    user: Annotated[UserVerifySchema, Depends(get_current_user)],
    service: ProfileService = Depends(ProfileService),
):
    return await service.change_email(data=data, user=user)

Разберём аргументы подробнее, чтобы даже новичкам было понятно, откуда всё берётся:

Почему используем **Annotated**? Так мы явно связываем схему UserVerifySchema и зависимость get_current_user. Это делает сигнатуру функции самодокументируемой: читающий код сразу видит, какой именно объект попадёт в переменную user.

После объявления этого маршрута смена e‑mail полностью готова. Протестируем чуть позже, когда реализуем аналогичный функционал для смены пароля.

Смена пароля

Переходим ко второму сценарию — смене пароля пользователя. Здесь уже не обойтись без дополнительных проверок:

1. Подтвердить текущий пароль. Прежде чем менять секрет, мы должны убедиться, что пользователь действительно знает свой старый пароль.
2. Захэшировать новый пароль. Сырые пароли нельзя хранить в базе данных. Мы используем надёжный алгоритм bcrypt , превращая строку в криптографический хэш.
3. Сохранить хэш в БД. После успешной проверки и хэширования новый пароль отправляется в таблицу user.

Если вы не сталкивались с термином хэширование: это одностороннее «перемешивание» строки. По хэшу невозможно восстановить исходный пароль, но можно проверить, совпадает ли он с тем, что ввёл пользователь. Подробнее о том, как именно работает шифрование паролей, мы разбирали в статьях «FastAPI 5. Приложение аутентификации и Pydantic схемы» и «FastAPI 6. Пользовательский сервис и маршруты регистрации».

Как и в разделе про изменение почты, начнём «с середины» — со схемы данных, а не с маршрута. Это позволит нам сразу пользоваться автодополнением при написании сервиса и роутера.

Схемы профиля — ChangePasswordRequest

Открываем schemas.py и добавляем новую Pydantic-схему ChangePasswordRequest.

from typing import Annotated

from pydantic import BaseModel, StringConstraints


class ChangePasswordRequest(BaseModel):
    old_password: Annotated[str, StringConstraints(min_length=8, max_length=128)]
    new_password: Annotated[str, StringConstraints(min_length=8, max_length=128)]

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

На этом подготовка схемы закончена. Далее мы реализуем метод получения хэшированного пароля пользователя из БД.

Менеджер профиля — получение хэша пароля

Нам предстоит извлечь из базы хэш текущего пароля пользователя, чтобы сравнить его с тем, что прислал пользователь. Для этого в managers.py, внутри класса ProfileManager, добавим метод get_user_hashed_password. Он принимает идентификатор пользователя user_id и возвращает строку-хэш.

Сначала открываем асинхронную сессию через async with self.db.db_session() as session. Далее формируем запрос с помощью select() из SQLAlchemy, выбрав лишь поле hashed_password у модели User, и добавляем фильтр .where(self.user_model.id user_id).

После выполнения запроса (await session.execute(query)) получаем объект Result. Метод scalar() вытаскивает первое значение первой строки — именно то, что нам нужно. Если запись не найдена, scalar() вернёт None.

import uuid
from sqlalchemy import select

async def get_user_hashed_password(self, user_id: uuid.UUID | str) -> str | None:
    async with self.db.db_session() as session:
        query = (
            select(self.user_model.hashed_password)
            .where(self.user_model.id == user_id)
        )
        result = await session.execute(query)
        return result.scalar()

Почему scalar(), а не scalars().first()? 
scalar() сразу возвращает одно-единственное поле из первой строки, поэтому короче и нагляднее, когда мы ожидаем ровно одно значение. Если бы нужно было получить несколько полей или строк, использовали бы scalars() или fetchall().

Далее мы реализуем сервис, который сравнит хэши старого пароля, создаст хэш для нового и передаст данные менеджеру БД.

Сервис профиля — бизнес-логика смены пароля

Открываем services.py и приводим сервис к новым требованиям.

__init__

В конструктор добавляем ещё одну зависимость — объект AuthHandler, который умеет сравнивать и хэшировать пароли. Как и менеджер, он передаётся через Depends.

from fastapi import Depends
from lkeep.apps.auth.handlers import AuthHandler
from lkeep.apps.profile.managers import ProfileManager


class ProfileService:
    def __init__(
        self,
        manager: ProfileManager = Depends(ProfileManager),
        handler: AuthHandler = Depends(AuthHandler),
    ) -> None:
        self.manager = manager
        self.handler = handler

Небольшое отступление о слоях. 
Идеально было бы вынести AuthHandler в общий пакет вроде core/security, чтобы избежать прямой зависимости между приложениями. Для учебного MVP это усложнило бы структуру, поэтому пока подключаем обработчик напрямую и оставляем рефакторинг «на потом».

change_password

Метод change_password принимает:

• data — объект ChangePasswordRequest с полями old_password и new_password;
• user — информацию об авторизованном пользователе (UserVerifySchema).

Алгоритм работы:

1. Получаем хэш текущего пароля из базы с помощью ProfileManager.
2. Сверяем его с тем, что прислал клиент (AuthHandler.verify_password).
3. Если хэши совпали, генерируем новый хэш и сохраняем его через update_user_fields.
4. Когда проверка не прошла, возвращаем ошибку 401 Unauthorized.

async def change_password(self, data: ChangePasswordRequest, user: UserVerifySchema) -> None | JSONResponse:  
    current_password_hash = await self.manager.get_user_hashed_password(user_id=user.id)  

    if await self.handler.verify_password(raw_password=data.old_password, hashed_password=current_password_hash):  
        hashed_password = await self.handler.get_password_hash(password=data.new_password)  
        await self.manager.update_user_fields(user_id=user.id, hashed_password=hashed_password)  
        return None  

    return JSONResponse({"error": "Invalid password"}, status_code=401)

Почему JSONResponse, а не HTTPException? 
Оба варианта допустимы. В примере используется JSONResponse, чтобы явно показать возвращаемый формат и статус-код. При желании можно заменить на HTTPException(status_code=401, detail="Invalid password") — результат для клиента будет тем же.

Маршруты профиля — смена пароля

Финишная прямая: добавим HTTP-маршрут, который свяжет всё, что мы уже написали, с внешним миром. Откройте routes.py и разместите под маршрутом смены e-mail ещё один эндпоинт.

Что делает декоратор

• указывает метод POST;
• задаёт путь /change-password;
• фиксирует успешный ответ как 200 OK.

@profile_router.post(
    "/change-password",
    status_code=status.HTTP_200_OK,
)
async def change_password(
    data: ChangePasswordRequest,
    user: Annotated[UserVerifySchema, Depends(get_current_user)],
    service: ProfileService = Depends(ProfileService),
) -> Response:
    return await service.change_password(data=data, user=user)

Функция-обработчик принимает:

• data — уже валидированный объект ChangePasswordRequest;
• user — данные авторизованного пользователя, которые извлекает зависимость get_current_user;
• service — экземпляр ProfileService, предоставленный через Depends.

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

Почему так лаконично? 
Все проверки, валидация и запись в базу данных уже инкапсулированы в сервисах и менеджерах. Роутер должен лишь принимать запрос, передавать данные глубже по слоям и отдавать клиенту ответ. Чем меньше кода в контроллерах, тем легче их читать и тестировать.

Регистрация роутера

Маршруты «profile» пока недоступны извне, потому что главный роутер проекта о них не знает. Подключим их в файле apps/__init__.py — там уже присутствует код для auth_router, добавим вторую строку для profile_router.

from fastapi import APIRouter

from lkeep.apps.auth.routes import auth_router       # ⬅︎ был раньше
from lkeep.apps.profile.routes import profile_router # ⬅︎ добавляем

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

apps_router.include_router(router=auth_router)
apps_router.include_router(router=profile_router)

Что здесь происходит:

• apps_router задаёт единый префикс /api/v1, поэтому конечные URL получаются вида 
  /api/v1/profile/change-email и /api/v1/profile/change-password.
• include_router «вкладывает» дочерние роутеры в общий, сохраняя их собственные префиксы и теги. 
  Благодаря этому Swagger-UI автоматически покажет новые эндпоинты в разделе profile.

Тестируем!

Запустите приложение командой

poetry run app

и откройте документацию Swagger-UI по адресу http://127.0.0.1:8000/docs.

На экране появятся все доступные маршруты:

Чтобы проверить операции смены почты и пароля, сначала авторизуйтесь (если в базе ещё нет учётной записи, зарегистрируйтесь).

Авторизация

Найдите секцию auth и раскройте эндпоинт авторизации, затем нажмите Try it out.

В появившихся полях укажите логин и пароль пользователя и нажмите Execute. Ниже Swagger-UI отобразит результат запроса. Сообщение «Вход успешен» означает, что вы получили токен и можете выполнять защищённые операции.

Смена почты

Перейдите к маршруту смены e-mail, снова нажмите Try it out.

Введите новый адрес в поле new_email и нажмите Execute.

Если в ответе пришёл пустой JSON со статус-кодом 200 OK, почта изменена. Убедиться в этом можно, заглянув в базу данных:

Смена пароля

Аналогично откройте маршрут смены пароля и нажмите Try it out.

Укажите текущий пароль в поле old_password и новый в new_password, после чего нажмите Execute.

Пустой ответ со статусом 200 OK означает, что в базе сохранён новый хэш пароля:

Для проверки обработчика ошибок повторите запрос, указав неверный текущий пароль. Swagger-UI вернёт сообщение об ошибке и статус 401 Unauthorized:

Заключение

У нас получилось аккуратно встроить в сервис два важных сценария — смену электронной почты и пароля — не нарушив общей архитектуры проекта. Мы:

• вынесли логику в отдельное приложение profile, чтобы придерживаться принципа единой ответственности;
• разделили код на router → service → manager, тем самым сохранив читаемость и упростив тестирование;
• показали, как безопасно обрабатывать пароли: проверка старого хэша, генерация нового, сохранение через транзакцию;
• добавили новые эндпоинты в главный роутер и убедились в их работе через Swagger-UI.

В следующей статье начнём писать приложение для хранения сокращённых ссылок. До финиша не далеко!