FastAPI 11. Хранение и сокращение ссылок

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

В этой статье мы с вами:

• Создадим модель для хранения ссылок в базе данных.
• Напишем четыре маршрута и всё необходимое к ним:
  • Маршрут получения полной ссылки по короткой.
  • Маршрут получения всех ссылок пользователя (для страницы в личном кабинете).
  • Маршрут создания ссылки.
  • Маршрут удаления ссылки.

Если вы попали на эту статью не с начала цикла, то рекомендую прочесть и другие статьи: Сервис на FastAPI. В нём мы изучаем FastAPI и пишем бэкэнд сервиса по сокращению ссылок.

Модель ссылок и миграции

Первым делом опишем модель для хранения ссылок в БД.

Класс модели ссылок

В пакете database/models создадим файл links.py и сразу его откроем.

В нём создаём класс Link, унаследованный от IDMixin, CreatedAtMixin и Base (именно в таком порядке, Base должен быть последним). Наследование даст нам поля id и created_at.

Далее нужно прописать три поля:

• full_link - строковый тип без указания длины. В нём будет храниться полная ссылка
• short_link - строковый тип с максимальной длиной в 12 символов. В нём будет храниться короткая ссылка
• owner_id - UUID тип. В нём будет храниться идентификатор пользователя со ссылкой на его запись в БД

Код класса:

from sqlalchemy import UUID, ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column

from lkeep.database.mixins.id_mixins import IDMixin
from lkeep.database.mixins.timestamp_mixins import CreatedAtMixin
from lkeep.database.models import Base


class Link(IDMixin, CreatedAtMixin, Base):
    full_link: Mapped[str] = mapped_column(String)
    short_link: Mapped[str] = mapped_column(String(12), unique=True, index=True)
    owner_id: Mapped[UUID] = mapped_column(ForeignKey("user.id", ondelete="CASCADE"))

Почему в full_link мы указываем просто String без длины?

Всё дело в том, как воспринимает этот тип база данных. Если указать, например, String(12), то в БД это будет VARCHAR(12), но если не указывать, то это будет TEXT - практически без ограничений по длине. Некоторые ссылки, особенно с множеством query-параметров, могут быть очень длинными.

Видимость для Alembic

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

Там уже есть наша базовая модель и модель пользователя. Добавляем ещё и модель ссылок.

Будет выглядеть так:

from lkeep.database.models.base import Base
from lkeep.database.models.links import Link
from lkeep.database.models.user import User


__all__ = ("Base", "User", "Link")

Создание и применение миграций

Осталось только создать и применить миграции. Для этого нужно выполнить две команды.

Первая создаёт миграцию:

poetry run alembic revision --autogenerate -m "Link Model"

Вторая применяет её к базе данных:

poetry run alembic upgrade head

Что такое Alembic?

Alembic - это инструмент для управления миграциями базы данных в SQLAlchemy. Он позволяет версионировать изменения схемы БД, откатывать их и синхронизировать между различными окружениями. Команда revision --autogenerate автоматически создаёт миграцию, сравнивая текущие модели с состоянием БД.

Подробнее в статье: FastAPI 4. Модель пользователя, миксины и Alembic

Ограничитель короткой ссылки

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

Обратите внимание! В описании БД указано, можно сказать, захардкожено число 12. Таким образом, если вы измените максимальную длину ссылки, скажем, на 15, то при вставке в БД будет ошибка. Значение не должно превышать то, что указано в БД. Это можно решить, например, задав для БД длину 255, но если будет использоваться только 8-16 символов, то это будет пустая трата памяти.

Откроем файл settings.py в пакете core и найдём там класс Settings.

Добавим в него новый параметр link_length:

class Settings(BaseSettings):
    ...
    link_length: int = 12

Можно задать значение по умолчанию, а актуальное значение прописать в .env.

Маршруты ссылок

Теперь перейдём к написанию маршрутов.

Подготовка

Прежде чем начать писать маршруты, нужно всё подготовить.

Первым делом в пакете apps создадим новый пакет links. В нём будет находиться логика работы со ссылками и обработка маршрутов.

Сразу создадим в нём все необходимые файлы:

• managers.py - файл с менеджером для работы с БД
• routes.py - файл с маршрутами
• schemas.py - файл с Pydantic-схемами
• services.py - файл с сервисом

managers.py

В этом файле пропишем класс LinksManager:

from fastapi import Depends

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


class LinksManager:
    def __init__(self, db: DBDependency = Depends(DBDependency)) -> None:
        self.db = db
        self.link_model = Link

Это основа класса. Такая же, как и в других слоях приложения.

routes.py

В этом файле создадим объект роутера ссылок:

from fastapi import APIRouter


links_router = APIRouter(prefix="/links", tags=["links"])

Прописываем, что у него будет маршрут /links/<эндпоинт> и группу тегов для Swagger.

Также, после того как его прописали, нужно его зарегистрировать. Для этого откроем файл __init__.py в пакете apps и добавим новый роутер:

from fastapi import APIRouter

from lkeep.apps.auth.routes import auth_router
from lkeep.apps.links.routes import links_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.include_router(router=links_router)

services.py

Также как и с менеджером, пропишем основу класса:

from fastapi import Depends

from lkeep.apps.links.managers import LinksManager


class LinksService:
    def __init__(self, manager: LinksManager = Depends(LinksManager)) -> None:
        self.manager = manager

Тут ничего нового.

schemas.py

Заранее создадим пять схем:

• BaseFullLink - базовая схема. Содержит поле full_link для наследования другими схемами
• DeleteLinkSchema - базовая схема. Содержит поле id. Используется в других схемах и маршруте удаления ссылки
• LinkSchema - полная схема объекта ссылки. Наследуется от BaseFullLink и DeleteLinkSchema для получения их полей
• GetLinkSchema и CreateLinkSchema - схемы, наследуемые от BaseFullLink. Используются в одноимённых маршрутах

Код классов:

import uuid
from datetime import datetime

from pydantic import BaseModel


class BaseFullLink(BaseModel):
    full_link: str


class DeleteLinkSchema(BaseModel):
    id: uuid.UUID


class LinkSchema(BaseFullLink, DeleteLinkSchema):
    short_link: str
    created_at: datetime


class GetLinkSchema(BaseFullLink):
    pass


class CreateLinkSchema(BaseFullLink):
    pass

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

Первый будет маршрут, который возвращает только значение full_link.

В файле routes.py создаём функцию get_link. Она принимает два аргумента: строку short_link и service через Depends. Возвращает функция объект GetLinkSchema или None, если не найдена ссылка.

Что такое Depends?

Depends - это реализация Dependency Injection (внедрения зависимостей) от FastAPI. Он позволяет автоматически создавать и передавать зависимости в функции маршрутов. В нашем случае FastAPI автоматически создаст экземпляр LinksService и передаст его в функцию.

Оборачиваем функцию в декоратор от роутера. Вызываем метод .get() и передаём аргументы: путь /get_link, response_model - возвращаемая схема и status_code.

Внутри функции вызываем метод get_link из сервиса.

from fastapi import Depends
from starlette import status

from lkeep.apps.links.schemas import GetLinkSchema
from lkeep.apps.links.services import LinksService


@links_router.get("/get_link", response_model=GetLinkSchema | None, status_code=status.HTTP_200_OK)
async def get_link(short_link: str, service: LinksService = Depends(LinksService)) -> GetLinkSchema | None:
    return await service.get_link(short_link=short_link)

Переходим в сервис. Там создаём метод get_link, принимающий только строку short_link.

В теле метода вызываем у менеджера метод .get_link().

from lkeep.apps.links.schemas import GetLinkSchema


async def get_link(self, short_link: str) -> GetLinkSchema | None:
    return await self.manager.get_link(short_link=short_link)

Переходим в менеджер и прописываем метод get_link. Он также принимает только строку short_link.

В теле метода открываем асинхронный контекстный менеджер с сессией базы данных. Затем в переменную query прописываем запрос: нам нужно значение поля full_link для записи, содержащей такой же short_link.

Затем выполняем запрос и, используя scalar_one_or_none, получаем готовый строковый объект, если запись найдена, в противном случае будет None.

После чего возвращаем или Pydantic-схему, или ничего.

from sqlalchemy import select

from lkeep.apps.links.schemas import GetLinkSchema


async def get_link(self, short_link: str) -> GetLinkSchema | None:
    async with self.db.db_session() as session:
        query = select(self.link_model.full_link).where(self.link_model.short_link  short_link)

        result = await session.execute(query)
        link = result.scalar_one_or_none()

        if link:
            return GetLinkSchema(full_link=link)

        return None

Маршрут готов!

Теперь, когда пользователь откроет ссылку вида https://lkeep.ru/qwerty123456, с фронтенда будет послан запрос на получение полной ссылки, и если она есть, фронтенд перенаправит пользователя на целевой сайт.

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

Результатом работы этого маршрута будет список (массив) с объектами ссылок. Они будут отображаться только пользователю, создавшему их.

В файле routes.py создадим новый маршрут get_user_links. Он принимает два аргумента: user - объект класса UserVerifySchema и service. Получение авторизованного пользователя описывалось в статье FastAPI 9. Logout и проверка авторизации.

В декораторе роутера выбираем метод .get(), прописываем маршрут /get_user_links, в качестве response_model будет список объектов LinkSchema и статус-код 200.

Внутри функции вызываем get_links у сервиса и передаём туда пользователя.

from typing import Annotated

from lkeep.apps.auth.depends import get_current_user
from lkeep.apps.auth.schemas import UserVerifySchema
from lkeep.apps.links.schemas import LinkSchema


@links_router.get("/get_user_links", response_model=list[LinkSchema], status_code=status.HTTP_200_OK)
async def get_user_links(
    user: Annotated[UserVerifySchema, Depends(get_current_user)], 
    service: LinksService = Depends(LinksService)
) -> list[LinkSchema]:
    return await service.get_links(user=user)

Переходим в services.py. Тут всё достаточно просто.

Создаём метод get_links, принимающий пользователя и вызывающий у менеджера get_links с аргументом user_id.

from lkeep.apps.auth.schemas import UserVerifySchema
from lkeep.apps.links.schemas import LinkSchema


async def get_links(self, user: UserVerifySchema) -> list[LinkSchema]:
    return await self.manager.get_links(user_id=user.id)

В менеджере примерно так же, как и в предыдущем методе.

Создаём метод get_links, принимающий user_id. В теле открываем контекстный менеджер.

В переменную query составляем запрос: получаем весь объект модели, если owner_id из модели совпадает с идентификатором авторизованного пользователя.

Выполняем запрос и, используя result.scalars().all(), получаем список объектов модели.

Затем возвращаем список валидированных объектов Pydantic, а если данных в БД не было, то вернётся пустой список.

import uuid

from lkeep.apps.links.schemas import LinkSchema


async def get_links(self, user_id: uuid.UUID) -> list[LinkSchema]:
    async with self.db.db_session() as session:
        query = select(self.link_model).where(self.link_model.owner_id  user_id)

        result = await session.execute(query)
        links = result.scalars().all()

        return [LinkSchema.model_validate(link, from_attributes=True) for link in links]

Готово!

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

Маршрут создания ссылки

Третьим маршрутом будет создание ссылки.

Начало такое же, как и у всех. Создаём функцию create_link, принимающую помимо пользователя и сервиса объект класса CreateLinkSchema в аргумент link_data. Прописываем в декораторе метод .post() и аргументы: маршрут /create_link, response_model - LinkSchema и статус 201 - Создан.

В теле вызываем метод сервиса create_link, передав в него данные и пользователя.

from lkeep.apps.links.schemas import CreateLinkSchema


@links_router.post("/create_link", response_model=LinkSchema, status_code=status.HTTP_201_CREATED)
async def create_link(
    link_data: CreateLinkSchema,
    user: Annotated[UserVerifySchema, Depends(get_current_user)],
    service: LinksService = Depends(LinksService),
) -> LinkSchema:
    return await service.create_link(link_data=link_data, user=user)

Переходим в сервис. Создаём функцию create_link, принимающую данные из маршрута.

В теле функции в переменную link_length получаем значение из настроек.

Затем открываем бесконечный цикл while True.

Внутри цикла в переменную short_link, используя метод .token_urlsafe() из встроенной в Python библиотеки secrets, генерируем произвольный набор символов и урезаем его до лимита (в нашем случае до 12 символов).

Почему используем secrets.token_urlsafe()?

Модуль secrets предназначен для генерации криптографически стойких случайных чисел, подходящих для управления секретными данными. Функция token_urlsafe() возвращает случайную URL-безопасную текстовую строку в кодировке Base64, что идеально подходит для коротких ссылок.

Далее открываем блок try-except. В блоке try возвращаем вызов метода create_link из менеджера, передав в него полную ссылку, идентификатор пользователя и сгенерированную короткую ссылку. В блоке except отлавливаем исключение IntegrityError, которое сигнализирует о наличии дубликата короткой ссылки. Там просто запускаем новую итерацию цикла while.

Суть в том, что если короткая ссылка уже есть в БД, то мы продолжаем генерировать новые, пока не сгенерируется уникальная, а если всё хорошо, то результат вернётся в маршрут и будет произведён выход из цикла. Но, учитывая что на 12-ти символах ОЧЕНЬ много вариантов, словить дубликаты будет крайне сложно.

import secrets

from sqlalchemy.exc import IntegrityError

from lkeep.core.settings import settings
from lkeep.apps.links.schemas import CreateLinkSchema


async def create_link(self, link_data: CreateLinkSchema, user: UserVerifySchema) -> LinkSchema:
    link_length = settings.link_length

    while True:
        short_link = secrets.token_urlsafe(link_length)[:link_length]
        try:
            return await self.manager.create_link(
                full_link=link_data.full_link, user_id=user.id, short_link=short_link
            )
        except IntegrityError:
            continue

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

Открыв контекстный менеджер, прописываем переменную query с запросом: вставить в модель указанные данные, а затем вернуть созданный объект модели.

Выполняем и фиксируем запрос, затем, используя result.scalar(), получаем созданный объект.

Возвращаем его в виде Pydantic-схемы.

from sqlalchemy import insert


async def create_link(self, full_link: str, user_id: uuid.UUID, short_link: str) -> LinkSchema:
    async with self.db.db_session() as session:
        query = (
            insert(self.link_model)
            .values(full_link=full_link, short_link=short_link, owner_id=user_id)
            .returning(self.link_model)
        )

        result = await session.execute(query)
        await session.commit()
        link = result.scalar()

        return LinkSchema.model_validate(link, from_attributes=True)

Готово!

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

Маршрут удаления ссылки

Последний маршрут - удаление ссылки.

Возвращаемся в файл с маршрутами и создаём новый delete_link. Он принимает уже привычные аргументы - сервис и пользователя, а также link_data - объект схемы DeleteLinkSchema с идентификатором удаляемой ссылки.

В декораторе роутера вызываем метод .delete() с маршрутом /delete_link и статус-кодом 204 - Без ответа.

В теле маршрута вызываем сервис, передав в него данные и пользователя.

from lkeep.apps.links.schemas import DeleteLinkSchema


@links_router.delete("/delete_link", status_code=status.HTTP_204_NO_CONTENT)
async def delete_link(
    link_data: DeleteLinkSchema,
    user: Annotated[UserVerifySchema, Depends(get_current_user)],
    service: LinksService = Depends(LinksService),
) -> None:
    await service.delete_link(link_data=link_data, user=user)

В сервисе также создаём метод delete_link.

Для того чтобы убедиться, что пользователь, инициировавший удаление, является создателем ссылки, а не "чуваком с Postman", сперва в переменную link_owner вызываем метод get_link_owner из менеджера, передав туда идентификатор ссылки. Там мы получим идентификатор создавшего её пользователя или None, если такой ссылки уже нет.

Затем в блоке if проверяем, что данные из БД получены и идентификатор пользователя совпадает с создателем ссылки. Если да, вызываем метод delete_link, передав в него идентификатор удаляемой ссылки.

Если же ссылки нет в БД или пользователь не совпал, то в блоке else поднимаем исключение HTTPException. Указываем статус-код 403 - Запрещено и пояснительный текст, что пользователь не совпал.

from fastapi import HTTPException

from lkeep.apps.links.schemas import DeleteLinkSchema


async def delete_link(self, link_data: DeleteLinkSchema, user: UserVerifySchema) -> None:
    link_owner = await self.manager.get_link_owner(link_id=link_data.id)

    if link_owner and link_owner  user.id:
        await self.manager.delete_link(link_id=link_data.id)
    else:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Wrong link owner")

Переходим в менеджера. Там надо создать два метода.

Первый get_link_owner похож на метод get_link, только без Pydantic-схемы. В нём делаем запрос на получение поля owner_id у записи с таким же id, как и у удаляемой ссылки.

Во втором delete_link в запросе вызываем удаление объекта таблицы по совпадающему идентификатору.

from sqlalchemy import delete


async def get_link_owner(self, link_id: uuid.UUID) -> uuid.UUID | None:
    async with self.db.db_session() as session:
        query = select(self.link_model.owner_id).where(self.link_model.id  link_id)

        result = await session.execute(query)
        return result.scalar_one_or_none()

async def delete_link(self, link_id: uuid.UUID) -> None:
    async with self.db.db_session() as session:
        query = delete(self.link_model).where(self.link_model.id  link_id)

        await session.execute(query)
        await session.commit()

Готово!

Пользователю в личном кабинете будут отображаться его ссылки с кнопками "удалить". Нажав на кнопку, будет отправлен запрос от фронтенда. У нас мы проверим, что пользователь, создавший ссылку, и запросивший удаление - один и тот же, после чего удаляем и возвращаем только 204-й статус-код.

Заключение

Вот так достаточно просто и быстро реализуется логика хранения, создания, выдачи и удаления ссылок.

Тут ещё много всего можно придумать, например, добавить редактирование или ещё что-то, но мы на этом закончим.

Можно считать, что первый MVP готов, теперь осталось сделать для него фронтенд.

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