services/fleet/api.py - dev - Arslan16/FleetHub

services/fleet/api.py
401 строка14 KB
Arslan16
Для agent-gateway написал тесты (cov 71%). В fleet добавил эндпоинт для проверки владения пользователем сервера
11 авг 2026, 01:16
11 авг 2026, 01:166701f0a
"""
Модуль API маршрутов сервиса Fleet.

Содержит HTTP-эндпоинты для управления серверами пользователей:
- получение структуры данных для создания сервера;
- создание новых серверов;
- получение списка серверов пользователя;
- получение полной информации о сервере;
- обновление данных сервера;
- удаление серверов.

Доступные маршруты:
    GET /api/create:
        Получение описания полей, необходимых для создания сервера.

    POST /api/create:
        Создание нового сервера пользователя.

    GET /api/list:
        Получение списка серверов текущего пользователя.

    POST /api/get-full-info:
        Получение полной информации о конкретном сервере.

    POST /api/update:
        Обновление данных существующего сервера.

    POST /api/delete:
        Удаление сервера пользователя.

Использует:
    - ServersManager для работы с данными серверов;
    - сервис авторизации для проверки пользователя;
    - SQLAlchemy AsyncSession через зависимости FastAPI;
    - Pydantic модели для валидации данных запросов.
"""
from fastapi import APIRouter, Request, Depends
from fastapi.responses import JSONResponse
from starlette import status

from database_utils import ServersManager
from dependencies import (
    auto_upsert_user,
    authenticate_user,
    get_servers_manager
)
from schemas import (
    UserContext,
    Server,
    ServerId,
    UpdateServer,
    server_model_json
)
from logger import logger
from models import ServerTable


api_rt: APIRouter = APIRouter(prefix="/api", dependencies=[Depends(auto_upsert_user)])
"Роутер отвечающий за API сервиса сервреров. CRUD операции"


@api_rt.get("/create")
async def create_server_get_handler(
    request: Request,
    user: UserContext = Depends(authenticate_user)
) -> JSONResponse:
    """
    Отправляет список полей и их типы необходимые для создания сервера

    Args:
        user (UserContext): информация о пользователе

    Returns:
        response (JSONResponse): Поля и их типы под ключом `fields`
    """
    logger.info(
        "Запрос полей, необходимых для создания сервера",
        {
            "request": logger.get_request_info(request),
            "user_id": user.id
        }
    )
    return JSONResponse(server_model_json)


@api_rt.post("/create")
async def create_server_post_handler(
    request: Request,
    data: Server,
    user: UserContext = Depends(authenticate_user),
    manager: ServersManager = Depends(get_servers_manager)
) -> JSONResponse:
    """
    Эндпоинт для создания сервера, принимает данные о сервере

    Args:
        request (Request): Текущий HTTP запрос
        data (Server): Pydantic модель с данными о сервере
        user (UserContext): Информация о пользователе
        manager (ServersManager): Менеджер для управления таблицей серверов в бд

    Returns:
        response (JSONResponse): Ответ с кодом 200 или 409 или 500
    """
    log_data: dict = {
        "request": logger.get_request_info(request),
        "data": data.model_dump(),
        "user_id": user.id
    }
    logger.info("Запрос на создание сервера", log_data)

    if await manager.exists_by_ipv4(user.id, data.ipv4)\
       or await manager.exists_by_ipv6(user.id, data.ipv6):
        logger.error("Попытка добавить сервер с существующим IP", log_data)
        return JSONResponse(
            {
                "ok": False,
                "error": "Server with such adress already exists in user's list"
            },
            status_code=status.HTTP_409_CONFLICT
        )

    result: ServerTable | None = await manager.create(
        user_id=user.id,
        domain=data.domain,
        ipv4=data.ipv4,
        ipv6=data.ipv6,
        os=data.os,
        agent_id=data.agent_id,
        status=data.status
    )
    if isinstance(result, ServerTable):
        logger.info("Сервер успешно создан!", {**log_data, "id": result.id})
        return JSONResponse(
            {
                "ok": True,
                "id": result.id
            },
            status_code=status.HTTP_200_OK
        )
    logger.error("Сервер создать не удалось!", log_data)
    return JSONResponse(
        {
            "ok": False,
            "error": "Enternal Server Error"
        },
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
    )


@api_rt.get("/list")
async def servers_list_get_handler(
    request: Request,
    user: UserContext = Depends(authenticate_user),
    manager: ServersManager = Depends(get_servers_manager)
) -> JSONResponse:
    """
    Отправляет список серверов пользователя в формате списка объектов json

    Args:
        request (Request): Текущий HTTP запрос
        user (UserContext): Информация о пользователе
        manager (ServersManager): Менеджер для управления таблицей серверов в бд

    Returns:
        response (JSONResponse): JSON со списком серверов пользователя\n\n\
        Поля 1 объекта: `domain`, `ipv4`, `ipv6`, `os`, `agent_id`, `status`, `created_at`

    """
    logger.info(
        "Запрос на полуечение списка серверов",
        {
            "request": logger.get_request_info(request),
            "user_id": user.id
        }
    )

    servers: list[dict] = await manager.get_dict_list(user.id)
    return JSONResponse(servers)


@api_rt.post("/get-full-info")
async def get_server_post_handler(
    request: Request,
    data: ServerId,
    user: UserContext = Depends(authenticate_user),
    manager: ServersManager = Depends(get_servers_manager)
) -> JSONResponse:
    """
    Эндпоинт для выдачи полной информации по конкретному серверу и его ID

    Args:
        request (Request): Текущий HTTP запрос
        user (UserContext): Информация о пользователе
        manager (ServersManager): Менеджер для управления таблицей серверов в бд

    Returns:
        response (JSONResponse): JSON с полной информацией о сервере\n\n\
    """
    log_data: dict = {
        "request": logger.get_request_info(request),
        "user_id": user.id,
        "server_id": data.id
    }
    logger.info("Запрос на получение полной информации о сервере", log_data)
    if not await manager.is_user_server_owner(user.id, data.id):
        logger.error("Пользователь попытался получить информацию о сервере, который ему не принадлежит", log_data)
        return JSONResponse(
            {
                "ok": False,
                "error": "User is not Server owner"
            },
            status_code=status.HTTP_403_FORBIDDEN
        )

    server: dict | None = await manager.get_server_by_id(data.id)
    if server is None:
        logger.error("Сервер с таким id не обнаружен!", log_data)
        return JSONResponse(
            {
                "ok": False,
                "error": "Server not found"
            },
            status_code=status.HTTP_404_NOT_FOUND
        )

    logger.info("Успешно отправлена информация о сервере", log_data)
    return JSONResponse(
        {
            "ok": True,
            "server": server
        },
        status_code=status.HTTP_200_OK
    )


@api_rt.post("/update")
async def update_server_post_handler(
    request: Request,
    data: UpdateServer,
    user: UserContext = Depends(authenticate_user),
    manager: ServersManager = Depends(get_servers_manager)
):
    """
    Эндпоинт для обновления информации о сервере

    Args:
        request (Request): Текущий HTTP запрос
        data (UpdateServer): Новые данные, на которые необходимо обновиться
        user (UserContext): Информация о пользователе
        manager (ServersManager): Менеджер для управления таблицей серверов в бд

    Returns:
        response (JSONResponse): статус успешно или нет был обновлен сервер
    """
    log_data: dict = {
        "request": logger.get_request_info(request),
        "user_id": user.id,
        "server_id": data.id
    }
    logger.info("Запрос на обновление информации о сервере", log_data)
    if not await manager.is_user_server_owner(user.id, data.id):
        logger.error("Пользователь попытался обновить информацию о сервере, который ему не принадлежит", log_data)
        return JSONResponse(
            {
                "ok": False,
                "error": "User is not Server owner"
            },
            status_code=status.HTTP_403_FORBIDDEN
        )

    result: bool = await manager.update_server(
        user_id=user.id,
        server_id=data.id,
        domain=data.domain,
        ipv4=data.ipv4,
        ipv6=data.ipv6,
        agent_id=data.agent_id
    )
    if result:
        logger.info("Информация о сервере успешно обновлена!", log_data)
        return JSONResponse(
            {
                "ok": True
            }
        )
    logger.error("Обновить информацию о сервере не удалось!", log_data)
    return JSONResponse(
        {
            "ok": False,
            "error": "Enternal Server Error"
        },
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
    )


@api_rt.post("/delete")
async def delete_server_post_handler(
    request: Request,
    data: ServerId,
    user: UserContext = Depends(authenticate_user),
    manager: ServersManager = Depends(get_servers_manager)
) -> JSONResponse:
    """
    Эндпоинт для удаления сервера

    Args:
        request (Request): Текущий HTTP запрос
        data (ServerId): Идентификатор сервера, который нужно удалить
        user (UserContext): Информация о пользователе
        manager (ServersManager): Менеджер для управления таблицей серверов в бд

    Returns:
        response (JSONResponse): статус успешно или нет был обновлен сервер
    """
    log_data: dict = {
        "request": logger.get_request_info(request),
        "user_id": user.id,
        "server_id": data.id
    }
    logger.info("Запрос на удаление информации о сервере", log_data)

    if not await manager.is_user_server_owner(user.id, data.id):
        logger.error("Пользователь попытался удлаить информацию о сервере, который ему не принадлежит", log_data)
        return JSONResponse(
            {
                "ok": False,
                "error": "User is not Server owner"
            },
            status_code=status.HTTP_403_FORBIDDEN
        )

    result: bool = await manager.delete_server(
        user_id=user.id,
        server_id=data.id
    )
    if result:
        logger.info("Сервер успешно удален!", log_data)
        return JSONResponse(
            {
                "ok": True
            }
        )
    logger.error("Удалить информацию о сервере не удалось!", log_data)
    return JSONResponse(
        {
            "ok": False,
            "error": "Enternal Server Error"
        },
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
    )


@api_rt.post("/check-ownership")
async def check_ownership_post_handler(
    request: Request,
    data: ServerId,
    user: UserContext = Depends(authenticate_user),
    manager: ServersManager = Depends(get_servers_manager)
) -> JSONResponse:
    """
    Эндпоинт для проверки является ли пользователь владельцем конкретного сервера

    Args:
        request (Request): Текущий HTTP запрос
        data (ServerId): Идентификатор сервера, который нужно удалить
        user (UserContext): Информация о пользователе
        manager (ServersManager): Менеджер для управления таблицей серверов в бд

    Returns:
        response (JSONResponse):
            - 200 `ok`: `True` Пользователь является владельцем
            - 403 `ok`: `False` плюс ключ `detail`. Пользователь не является владельцем сервера
    """
    log_data: dict = {
        "request": logger.get_request_info(request),
        "user_id": user.id,
        "server_id": data.id
    }
    logger.info("Запрос на проверку владения пользователем сервера", log_data)
    if await manager.is_user_server_owner(user.id, data.id):
        logger.info("Пользователь владеет сервером", log_data)
        return JSONResponse(
            {
                "ok": True
            },
            status_code=status.HTTP_200_OK
        )
    logger.info("Пользователь не является владельцем сервера", log_data)
    return JSONResponse(
        {
            "ok": False,
            "detail": "Forbidden"
        },
        status_code=status.HTTP_403_FORBIDDEN
    )


__all__: list[str] = [
    "api_rt"
]