app/core/dependencies.py - master - outthevoid/python_microblogging_service_project

app/core/dependencies.py
87 строк3 KB
VladimirChukavin
Revert "Вернул проект в исходное состояние."
07 апр 2026, 14:01
07 апр 2026, 14:017531b60
from typing import Annotated

from fastapi import Depends, Header
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AuthenticationError
from app.core.session import get_db_session
from app.models.user_model import User
from app.repositories.user_repo import UserRepository

DBSession = Annotated[AsyncSession, Depends(get_db_session)]
ApiKeyHeader = Annotated[str | None, Header()]


async def get_api_key(api_key: ApiKeyHeader = None) -> str | None:
    """
    Возвращает значение API-ключа из заголовка запроса.

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

    :param api_key: Значение API-ключа из заголовка запроса.
    :type api_key: str | None

    :returns: API-ключ или ``None``, если заголовок отсутствует.
    :rtype: str | None
    """

    return api_key


async def get_current_user_by_api_key(
    session: AsyncSession,
    api_key: str | None,
) -> User:
    """
    Возвращает пользователя, соответствующего переданному API-ключу.

    Функция проверяет наличие API-ключа, выполняет поиск пользователя через
    репозиторий и выбрасывает исключение аутентификации, если ключ отсутствует
    или пользователь не найден.

    :param session: Асинхронная сессия базы данных.
    :type session: sqlalchemy.ext.asyncio.AsyncSession
    :param api_key: API-ключ пользователя.
    :type api_key: str | None

    :raises app.core.exceptions.AuthenticationError: Если API-ключ отсутствует
        или пользователь с таким ключом не найден.

    :returns: Пользователь, связанный с указанным API-ключом.
    :rtype: app.models.user_model.User
    """

    if not api_key:
        raise AuthenticationError("Требуется API ключ")

    repository = UserRepository(session)
    user = await repository.get_by_api_key(api_key=api_key)
    if user is None:
        raise AuthenticationError("Пользователь с таким API ключем не найден")

    return user


async def get_current_user(
    session: DBSession, api_key: Annotated[str | None, Depends(get_api_key)]
) -> User:
    """
    Возвращает текущего аутентифицированного пользователя.

    Функция использует API-ключ, извлечённый из заголовка запроса, и делегирует
    проверку пользователя функции поиска по API-ключу.

    :param session: Сессия базы данных.
    :type session: DBSession
    :param api_key: API-ключ пользователя, полученный через зависимость.
    :type api_key: str | None

    :returns: Текущий аутентифицированный пользователь.
    :rtype: app.models.user_model.User
    """

    return await get_current_user_by_api_key(session=session, api_key=api_key)


CurrentUser = Annotated[User, Depends(get_current_user)]