Sobes.tech
Ассистент
Ресурсы
Подготовка
Ресурсы
Резюме
Работа
Ресурсы
Документы
Назад к вопросам
Middle
2

from fastapi import FastAPI, UploadFile, Response from fastapi.responses import StreamingResponse from pydantic import BaseModel, field_validator import base64 app = FastAPI() class PdfSchema(BaseModel): filename: str content_type: str data: str # base64 pdf 10мб @app.get("/data", response_model=PdfSchema) async def get_data(): return await service.get_big_data() @app.get("/download") async def download(): with open("big.pdf", "rb") as f: data = f.read() return Response(content=data, media_type="application/pdf") def heavy_pdf_parse(data: bytes) -> dict: # CPU-heavy логика return {"size": len(data)} @app.post("/parse") async def parse(file: UploadFile): content = await file.read() result = heavy_pdf_parse(content) return result @app.get("/users") async def get_users(): users = await repo.get_users() result = [] for u in users: posts = await repo.get_posts(u["id"]) result.append({"user": u, "posts": posts}) return result @app.get("/download-two") async def download_two(): def iterfile(): with open("big.pdf", "rb") as f: yield from f return StreamingResponse(iterfile(), media_type="application/pdf") class service: @staticmethod async def get_big_data(): return { "filename": "file.pdf", "content_type": "application/pdf", "data": base64.b64encode(b"fake").decode(), } class repo: @staticmethod async def get_users(): return [{"id": 1}, {"id": 2}] @staticmethod async def get_posts(user_id: int): return [{"id": 1, "user id": user_id}]

Компании, где спрашивали
СБЕРСБЕР

Ответ от нейросети

sobes.tech AI

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

  1. Передача больших PDF в base64 в JSON (PdfSchema):

    • Кодирует PDF в base64 и передаёт как строку. Это неэффективно для больших файлов (10 Мб и больше), так как base64 увеличивает размер примерно на 33%.
    • Лучше отдавать файлы напрямую через StreamingResponse, как в /download и /download-two.

  2. Отдача файла через Response и StreamingResponse:

    • В /download файл читается целиком в память, что плохо для больших файлов.
    • В /download-two используется генератор iterfile(), который читает файл по частям и отдаёт потоково — это более оптимально.

  3. CPU-heavy функция heavy_pdf_parse:

    • Если парсинг занимает много CPU, стоит рассмотреть вынесение этой логики в отдельный воркер или использовать асинхронные очереди, чтобы не блокировать event loop.

  4. Асинхронные запросы в /users:

    • В цикле для каждого пользователя вызывается await repo.get_posts(u["id"]), что последовательно и может быть медленно.
    • Лучше запускать все запросы параллельно через asyncio.gather для ускорения:
import asyncio

@app.get("/users")
async def get_users():
    users = await repo.get_users()

    async def get_user_posts(u):
        posts = await repo.get_posts(u["id"])
        return {"user": u, "posts": posts}

    result = await asyncio.gather(*(get_user_posts(u) for u in users))
    return result
  1. Валидация входных данных:
    • В коде нет валидации входящих данных, например, для /parse можно добавить Pydantic-модель или проверку типа файла.

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

ПредыдущийСледующий