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}]

The provided code has several important points worth discussing:

1. Passing large PDFs in base64 within JSON (PdfSchema):

   • Encodes the PDF in base64 and transmits it as a string. This is inefficient for large files (10 MB and above), as base64 increases the size by about 33%.
   • It's better to serve files directly via StreamingResponse, as in /download and /download-two.

2. Serving files via Response and StreamingResponse:

   • In /download, the file is read entirely into memory, which is bad for large files.
   • In /download-two, a generator iterfile() is used, which reads the file in parts and streams it — this is more optimal.

3. CPU-heavy function heavy_pdf_parse:

   • If parsing takes a lot of CPU, consider moving this logic to a separate worker or using asynchronous queues to avoid blocking the event loop.

4. Asynchronous requests in /users:

   • In the loop, for each user, await repo.get_posts(u["id"]) is called, which is sequential and can be slow.
   • It's better to run all requests in parallel using asyncio.gather to speed up:

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

5. Input data validation:
   • The code lacks validation of incoming data; for example, for /parse, a Pydantic model or file type check could be added.

Overall, the code demonstrates basic techniques for working with FastAPI, but for production, it’s advisable to improve handling of large files, request parallelism, and input validation.