Использовал ли Grafana и Prometheus в связке?
Как правильно обеспечить идемпотентность при обработке пачки документов из Kafka, если сервис может упасть в середине обработки?
Что такое Middleware в FastAPI и зачем оно нужно?
Как работает EventLoop в Python? Чем асинхронность отличается от многопоточности и многопроцессорности?
Что такое GIL в Python? Почему он мешает в многопоточности, но не мешает в асинхронности?
Как в FastAPI запустить синхронный блокирующий код (например, вычисление на 1 секунду) по ручке, не блокируя EventLoop?
Как параллельно сделать 10 запросов к внешнему API и вернуть результат?
Что произойдёт в asyncio.gather(), если в одной из задач возникнет ошибка? Как сделать, чтобы исключения не терялись?
Как ограничить параллелизм при 10 запросах к API, которое принимает максимум 3 одновременных запроса?
Чем Semaphore отличается от Lock в asyncio?
Что такое контекстный менеджер в Python? Зачем нужен? Как написать свой?
Чем декоратор отличается от генератора? Когда лучше использовать генератор вместо списка?
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}]
Комфортно ли тебе оформление по самозанятости или ИП?
Как скоро готов выйти на проект, если он понравится?
Как разделены границы между микросервисами? Что делает каждый сервис?
Работали ли с Kubernetes? Какие инструменты деплоя использовали?
За сколько работает добавление в начало и конец списка в Python?
Что такое GIL?
Какие тестовые фреймворки есть в Python? Вспомогательные библиотеки?