Техпрогон
ВойтиНачать бесплатно
Код-ревью Middle FastAPI и Pydantic спрашивают часто обновлено

Код-ревью эндпоинта FastAPI

Вопрос

Найдите проблемы в эндпоинте загрузки профиля с внешним сервисом.

import requests
from fastapi import FastAPI
from sqlalchemy.orm import Session
from .db import engine
from .models import User

app = FastAPI()
session = Session(engine)


@app.get('/users/{user_id}')
async def get_user(user_id: int):
    user = session.get(User, user_id)
    avatar = requests.get(
        f'https://avatars.example.com/{user.email}'
    ).json()
    return {
        'id': user.id,
        'email': user.email,
        'password_hash': user.password_hash,
        'avatar': avatar,
    }

Коротко

Исправленный вариант: Сессия берётся из зависимости, клиент из lifespan, ответ проходит через модель, внешний сервис не может уронить или подвесить эндпоинт.

Что должен найти ревьюер

  1. Ошибкастрока 8
    Одна глобальная сессия на все запросы: общий кеш объектов и транзакция, гонки между запросами; сессия на запрос через зависимость с yield
  2. Производительностьстрока 13
    Синхронный запрос к БД внутри async def блокирует цикл событий; нужен AsyncSession или обычный def
  3. Ошибкастрока 13
    Нет обработки отсутствующего пользователя: None.email даст 500 вместо 404
  4. Производительностьстрока 14
    Синхронный requests.get в async def блокирует весь процесс; нужен httpx.AsyncClient из lifespan
  5. Ошибкастрока 15
    Нет таймаута и обработки ошибок внешнего сервиса: зависший ответ подвесит запрос
  6. Безопасностьстрока 17
    В ответ уходит password_hash и email без проверки прав; нужен response_model с безопасными полями и авторизация

Разбор

Исправленный вариант:

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    avatar: dict | None = None

@app.get('/users/{user_id}', response_model=UserOut)
async def get_user(
    user_id: int,
    session: Annotated[AsyncSession, Depends(get_session)],
    viewer: Annotated[User, Depends(current_user)],
    request: Request,
):
    user = await session.get(User, user_id)
    if user is None:
        raise HTTPException(404)
    try:
        r = await request.app.state.http.get(f'https://avatars.example.com/{user.id}', timeout=2)
        avatar = r.json() if r.is_success else None
    except httpx.HTTPError:
        avatar = None
    return UserOut(id=user.id, avatar=avatar)

Сессия берётся из зависимости, клиент из lifespan, ответ проходит через модель, внешний сервис не может уронить или подвесить эндпоинт.

Ответьте сами, а не прочитайте

В тренажёре вопрос попадёт в повторения и вернётся, когда начнёт забываться. Плюс билеты с зачётом: 20 вопросов, не больше 2 ошибок.

Потренироваться бесплатно

Предыдущий вопрос

Зачем response_model

Следующий вопрос

Подводные камни async в FastAPI

Урок по теме

FastAPI и Pydantic: маршруты, Depends, async и lifespan

Ещё по теме «FastAPI и Pydantic»

Все вопросы модуля