Если вы здесь, то вероятно уже знаете, что такое FastAPI. Это простой в понимании, легковесный веб-фреймворк изначально созданный для создания API. Работает он с помощью ASGI-сервера, о котором можно почитать во многих статьях, здесь же мы только кратко затронем его.

И без лишних слов, сегодня, как ясно из названия, мы будем писать FastAPI. В двух словах, он построен на Starlette, который работает с ASGI, и на Pydantic, который позволяет производить автоматическую валидацию получаемых данных. FastAPI же очень удобная оболочка над ними.

Что в статье

1. Рассмотрим приложение FastAPI, которое мы по итогу захотим запустить.

2. Напишем FastAPI, с рабочим названием myfastapi.

3. Запустим и проверим изначальное приложение написаное на fastapi с помощь myfastapi.

Что хотим написать

Все начинается с экземпляра главного класса FastAPI, который в свою очередь, во-первых наследуются от класса Starlette, который и выполняет всю основную работу по разрешению запросов, во-вторых создает декораторы, которые отправляют роуты в класс Starlette. Эти роуты хранятся там до запроса.

Когда происходит запрос к хосту на котором запущено приложение FastAPI, работая с ASGI фреймворком, приложение получает три параметра Scope, Receive и Send. Для работы ASGI приложению, точнее классу, нужен всего один метод __call__. Данный встроенный метод в классе FastAPI реализован лишь в виде вызова super().__call__, то есть класса Starlette. Но пока Starlette нас не интересует, до него мы еще дойдем.

В этой статье мы создадим функциональность, которая с помощью запуска ASGI сервера используя uvicorn main:app --reload, позволит работать следующему приложению

from typing import Union
from fastapi import FastAPI
from starlette.responses import JSONResponse
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None


app = FastAPI()


@app.post("/items/{item_id}")
async def create_item(item: Item, item_id: int):
    return JSONResponse({"item": item.dict(), "item_id": item_id})

@app.get("/")
async def homepage():
    return JSONResponse({'hello': 'world'})

@app.get("/get_items/{item_id}")
async def read_item(item_id: int):
    print("item_id", item_id)
    return {"item_id": item_id}

Для этого нам сделать, во-первых, декораторы, которые будут оборачивать роуты с их данными и отправлять в класс, который будет их обрабатывать, во-вторых, получения параметров пути, параметров запроса, тела запроса и валидацию на соответствие типам, если они используются, будь то встроенные типы, типы из typing или даже pydantic, что и является основной чертой FastApi, автоматической проверкой типов получаемых данных. 

Далее нам надо будет реализовать распределение запросов, то есть какой endpoint использовать при, например, GET запросе с путем “/items”. И в конце статьи, мы захотим сделать класс Response, который будет отлично работать в качестве ответа для ASGI сервера.

Конечно же, мы не будем использовать всю функциональность в рассматриваемых классах и методах, так как они могут охватывать большое количество путей использования. Поэтому будем использовать только необходимое, ведь когда мы это сделаем, то постепенно уже можно будет изучать и добавлять новую функциональность, доходя до реализации близкой к FastAPI. Если мы начнем охватывать все сразу, многое будет вызывать вопросы, но идя постепенно, код будет становититься все более понятным.

Пишем модуль myfastapi

Начнем с малого и самого простого - класс FastApi. Декораторы.

Как вы помните, инициализация простого приложения FastAPI, а возможно и большинства происходит следующим вызовом:

app = FastAPI()

Так как мы не будем рассматривать дебаггинг и OpenAPI здесь, то в метод __init__ мы просто добавим версию нашего приложения.

myfastapi::applications::FastApi::__init__

from typing import Callable
from starlette import Starlette
from myfastapi.routing import ApiRouter
 
class FastApi(Starlette):
    def __init__(
        self,
        version: str = "0.1.0"
    ) -> None:
        self.version = version
	  self.router: APIRouter = APIRouter()

Да, это все, что нам нужно здесь.

Сейчас посмотрим на то, что нас действительно интересует в этом классе - декораторы.

Мы видим класс ApiRouter, который мы будем использовать для создания декораторов, но об этом чуть позже.

На самом деле, как вы видите, наш класс FastApi не такой уж и простой в функциональности, так как он наследуется от класса Starlette, и это важно. На самом деле всю основную работу выполняет именно Starlette, а FastAPI просто удобная для использования оболочка над ним, но с довольно полезным функционалом и возможностью создания некоторого вида архитектуры для нашего приложения.

Итак, посмотрим на два метода get и post, которые используется в виде декораторов.

myfastapi::applications::FastApi::get

def get(
        self,
        path: str,
) -> Callable[..., Any]:
        return self.router.get(path)

myfastapi::applications::FastApi::post

def post(
        self,
        path: str,
) -> Callable[..., Any]:
        return self.router.post(path)

В двух словах, основной класс FastApi, который мы используем для создания приложения имеет под собой такой простой вызов методов класса ApiRouter.

Все, что мы передаем это путь, который будет запускать данный endpoint.

Становится очевидно, по тому, какой тип возврата указан у методов, что именно следующий вызов self.router.post(path) возвращает декоратор. Мы перейдем к классу ApiRouter уже скоро.

Так, помните, что для запуска ASGI приложения, необходимо, что класс был вызываемым, то есть имел метод __call__, и принимал три параметра Scope, Receive, Send. Где Scope отвечает за все метаданные и параметры пути, Receive хранит в себе тело запроса, а Send используется уже в самом конце обработки запроса для отправки заголовков и json ответа.

myfastapi::applications::FastApi::__call__

async def __call__(
self, scope: Scope, receive: Receive, send: Send
) -> None:
        await super().__call__(scope, receive, send)

Ага. Он просто вызывает класс Starlette, передавая ему полученные параметры.

Собственно, если использовать только два метода запроса get и post, так как для других методов используется идентичный подход, то класс FastApi для нашей библиотеки myfastapi, готов. Посмотрим на весь код вместе.

myfastapi::applications::FastApi

from typing import Callable
from starlette import Starlette
from myfastapi.routing import APIRouter
 
class FastApi(Starlette):
    def __init__(
        self,
        version: str = "0.1.0"
    ) -> None:
        self.version = version
    def get(
        self,
        path: str,
    ) -> Callable[..., Any]:
        return self.router.get(path)
    def post(
        self,
        path: str,
    ) -> Callable[..., Any]:
        return self.router.post(path)
    
    async def __call__(
        self, scope: Scope, receive: Receive, send: Send
    ) -> None:
        await super().__call__(scope, receive, send)

Точно, именно этот класс и будет использоваться для успешной инициализации нашего FastApi приложения.

Теперь пойдем дальше и перейдем в следующий файл routing.py, в котором у нас будет всего два класса ApiRouter и ApiRoute. Очевидно у них происходит определенное взаимодействие. ApiRouter действительно создает декораторы для методов запроса, возвращает из этих декораторы роуты, который представлены в виде класса ApiRoute, который в свою очередь содержит метод для проверки пути запроса и вызова endpoint если путь к экземпляре соответствующий, и по итогу, созданные из декораторов роуты или ApiRoute’s передаются в основной список роутов, в класс Router библиотеки starlette. И на этом работа FastApi с роутами заканчивается. А то, как проходить через них, какие методы вызывать у ApiRoute для соответствующих роутов остается на starlette.

Опять же, мы не будем лишний раз усложнять, и все добавляем постепенно. ApiRouter является классом для создания декораторов для функций, обрабатывающих запросы, и для передачи роутов к список, который хранится в starlette классе Routing, которые представлены классом ApiRoute состоящим из метода, пути, параметров и самой функции, то есть ApiRoute является оболочкой для определенного запроса.

Поэтому при инициализации ApiRouter мы без аргументов инициализируем класс Router от которого он наследуется, и просто добавим неинициализированный ApiRoute, который мы рассмотрим чуть позже, в переменную класса, для дальнейшей инициализации и использования при создании декоратора для определенного запроса.

myfastapi::routing::ApiRouter

from typing import Callable
from starlette.routing import Router, Any
 
class APIRouter(Router):
    def __init__(self) -> None:
        super().__init__()
        self.route_class = ApiRoute

Итак, следующий метод self.add_api_route будет принимать во-первых путь, который мы передадим в декоратор, например, @app.post("/items/"), во-вторых endpoint, то есть функцию, которая обрабатывает данные запрос, например:

async def create_item(item: Item):
    return JSONResponse(item.dict())

После чего эти данные будут использоваться для создания экземпляра класса ApiRoute, используя self.route_class, который будет сразу передаваться в список self.routes, находящийся в классе Routing (starlette).

myfastapi::routing::ApiRouter::add_api_route

def add_api_route(
        self,
        path: str,
        endpoint: Callable[..., Any],
	  method: str
) -> None:
	route = self.route_class(
    path, 
    endpoint=endpoint, 
    method=method
)
	self.routes.append(route)

Собственно это и есть главная функциональность данного класса. Теперь осталось сделать только декораторы для методов post и get (помните, что мы используем только их в этой статье?).

То есть, возвращаясь к классу FastApi, нам нужно сделать возможным следующее:

myfastapi::applications::FastApi::get

def get(
      self,
      path: str,
) -> Callable[..., Any]:
      return self.router.get(path)

И как вы догадываетесь, это очень просто:

myfastapi::routing::ApiRouter::get

def get(self, path: str) -> Callable[[Callable[..., Any]], [Callable[..., Any]]:
        def decorator(func: [Callable[..., Any]) -> [Callable[..., Any]:
	      self.add_api_route(path, func, method=”get”)
		return func
	  return decorator

Ничего необычного, простой декоратор, где func это наш endpoint для get запроса, и мы просто передаем путь и функцию для него в метод self.add_api_route, который мы рассмотрели выше. То есть таким простым способом мы регистрируем наши роуты в ASGI приложении.

И тоже самое для метода post

myfastapi::routing::ApiRouter::post

def post(self, path: str) -> Callable[[Callable[..., Any]], [Callable[..., Any]]:
        def decorator(func: [Callable[..., Any]) -> [Callable[..., Any]:
	      self.add_api_route(path, func, method=”post”)
		return func
	  return decorator

Мы также передаем и метод запроса, который используется для вызова функции для определенного пути, но в виде строки, хотя мы также могли бы привязать одну функцию для определенного пути для нескольких методов запроса. Просто будем иметь это ввиду.

Ну а сейчас, того, что есть в этом классе нам достаточно, посмотрим на весь получившийся класс:

myfastapi::routing::ApiRouter

from typing import Callable
from starlette.routing import Router, Any
 
class APIRouter(Router):
    def __init__(self) -> None:
        super().__init__()
        self.route_class = ApiRoute

    def add_api_route(
        self,
        path: str,
        endpoint: Callable[..., Any],
	  method: str
) -> None:
	route = self.route_class(
    path, 
    endpoint=endpoint, 
    method=method
)
	self.routes.append(route)

    def get(self, path: str) -> Callable[[Callable[..., Any]], [Callable[..., Any]]:
        def decorator(func: [Callable[..., Any]) -> [Callable[..., Any]:
	      self.add_api_route(path, func, method=”get”)
		return func
	  return decorator
 
    def post(self, path: str) -> Callable[[Callable[..., Any]], [Callable[..., Any]]:
        def decorator(func: [Callable[..., Any]) -> [Callable[..., Any]:
	      self.add_api_route(path, func, method=”post”)
		return func
	  return decorator

Да, так просто.

Теперь мы идем дальше, и уже начинается что-то интересное. Сейчас мы рассмотрим класс ApiRoute. Данный класс наследуется от starlette класса Route. И данный класс интересен нам тем, что он не просто является оболочкой, а переписывает поведение наследуемого класса. Вы наверное помните, что библиотека FastApi делает автоматическую валидацию типов для получаемых данных. Библиотека Starlette не поддерживает такую функциональность в своей основе, поэтому если бы вы использовали Starlette напрямую вместо FastApi, то вы могли бы присвоить своей функцию любую сигнатуру, то есть входящие переменные, присвоить им типы, но если в запросе могли бы прийти совершенно другие данные, и ошибка у вас возникла бы только при использовании заданных вами переменных функции, которая обрабатывает входящий запрос. FastApi же выдаст вам ошибку еще на этапе обработки полученных данных самой библиотекой, поэтому ваша функция даже не начнет работать, если входящие данные не соответствующие.

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

myfastapi::routing::ApiRoute

from starlette.routing import request_response
 
class ApiRoute(routing.Router):
    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        method: str
    ) -> None:
        self.path = path
        self.endpoint = endpoint
        self.method = method
        assert callable(endpoint), "An endpoint must be a callable"
        self.dependant = get_dependant(path=self.path, call=self.endpoint)
        self.app = request_response(get_request_handler())

При инициализации ApiRoute мы видим несколько дополнительных функций, где get_dependant ключевая для нас сейчас, так как ее вызов зависимости для роута в виде параметров пути, запроса и тела, к которым присваиваются названия параметров, типы, является ли параметр обязательным, и дефолтное значение если нет. В двух словах, FastApi работает таким образом, что если вы не укажите дефолтное значение, при этом поставите тип Optional из typing, и не передадите этот параметр при вызове, то возникнет ошибка. Чтобы сделать параметр необязательным, вам нужно передать дефолтное значение.

Сейчас мы рассмотрим get_dependant, а к переменной self.app вернемся после.

myfastapi::dependencies::utils::get_dependant

def get_dependant(
    *,
    path: str,
    call: Callable[..., Any],
) -> None:
    path_param_names = get_path_param_names(path)
    endpoint_signature = inspect.signature(call)
    signature_params = endpoint_signature.parameters
    dependant = Dependant(call=call, path=path)
    for param_name, param in signature_params.items():
        param_field = get_param_field(
            param=param, param_name=param_name
        )
        if param_name in path_param_names: dependant.path_params.append(param_field)
        elif (lenient_issubclass(param_field.type_, (list, set, tuple, dict)) or
        lenient_issubclass(param_field.type_, BaseModel)
        ): dependant.body_params.append(param_field)
        else: dependant.query_params.append(param_field)

Пойдем по порядку. get_dependant принимает путь и endpoint запроса. get_path_param_names просто возвращает нам имена параметров, которые являются частью пути:

myfastapi::utils::get_path_param_names

def get_path_param_names(path: str) -> Set[str]:
    return set(re.findall("{(.*?)}", path))

inspect.signature позволяет получить сигнатуру функции, то есть ее параметры с названием, типом и дефолтным значением. Класс Dependant просто хранит все параметры, которые в дальнейшем при вызове определенного роута будут использоваться для валидации полученных параметров.

myfastapi::dependencies::models::Dependant

from typing import Any, Callable, List, Optional
from pydantic.fields import ModelField
 
class Dependant:
    def __init__(
        self,
        *,
        path_params: Optional[List[ModelField]] = None,
        query_params: Optional[List[ModelField]] = None,
        body_params: Optional[List[ModelField]] = None,
        call: Optional[Callable[..., Any]] = None,
        path: Optional[str] = None
    ) -> None:
        self.path_params = path_params or []
        self.query_params = query_params or []
        self.body_params = body_params or []
        self.call = call
        self.path = path

Вы можете видеть, что параметры хранятся в виде класса ModelField библиотеки pydantic. В документации FastApi написано, он основан на pydantic, что имеет смысл, так как вся валидация параметров, которые делает FastApi происходит с помощью pydantic. Именно мы поэтому мы будем оборачивать наши параметры в ModelField.

Итак, возвращаясь к get_dependant, для каждого параметра endpoint функции, который был получен с помощью выявления сигнатуры, происходит цикл, в котором параметр сначала становится инстансом класса ModelField

param_field = get_param_field(
            param=param, param_name=param_name
        )

после чего, выявляя, что это за параметр, добавляется в класс Dependant:

if param_name in path_param_names: dependant.path_params.append(param_field)
elif (lenient_issubclass(param_field.type_, (list, set, tuple, dict)) or
        lenient_issubclass(param_field.type_, BaseModel)
): dependant.body_params.append(param_field)
else: dependant.query_params.append(param_field)

Функция get_param_field возвращает инстанс ModelField для параметра, и выглядит она следующим образом:

myfastapi::dependencies::utils::get_param_field

from pydantic import BaseConfig
from pydantic.fields import ModelField, Undefined
 

def get_param_field(
    param: inspect.Parameter,
    param_name: str
) -> ModelField:
    default_value: Any = Undefined
    if not param.default == param.empty: default_value = param.default
    required = True
    if default_value is not Undefined: required = False
    annotation: Any = Any
    if not param.annotation == param.empty:
        annotation = param.annotation
    field = ModelField(
        name=param_name,
        type_=annotation,
        default=default_value,
        class_validators=None,
        required=required,
        model_config=BaseConfig,
    )
     
    return field

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

При проверке, в какой список класса Dependant добавить данный инстанс параметра:

if param_name in path_param_names: dependant.path_params.append(param_field)
elif (lenient_issubclass(param_field.type_, (list, set, tuple, dict)) or
        lenient_issubclass(param_field.type_, BaseModel)
): dependant.body_params.append(param_field)
else: dependant.query_params.append(param_field)

Если название параметра в path_param_names, то есть он является частью пути, то он добавляется в path_params, если параметр является подклассом одной из четырех коллекций или BaseModel от pydantic, например:

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None

то данный параметр добавляется в список body_params, и соответственно, если параметр не является ни частью пути, ни тела запроса, то он добавляется в query_params и расматривается как параметр пути, который указывается после “?” в запросе.

Отлично! Наши зависимости разрешены, и мы можем запускать приложение в режиме --reload. Они будут сидеть и ждать своей очереди на использование до запроса, который бы вызывал endpoint к которому вместе с путем определенный Dependant и относится.

Все, что нам осталось сделать, это добавить методы для обработки запросов, и для этого мы возвращаемся к классу ApiRoute:

myfastapi::routing::ApiRoute

from starlette.routing import request_response
 
class ApiRoute(routing.Router):
    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        method: str
    ) -> None:
        self.path = path
        self.endpoint = endpoint
        self.method = method
        assert callable(endpoint), "An endpoint must be a callable"
        self.dependant = get_dependant(path=self.path, call=self.endpoint)
        self.app = request_response(get_request_handler(dependant=self.dependant))

Сейчас нам надо вспомнить то, о чем мы говорили в контексте ASGI приложения.

Чтобы обработать запрос ASGI приложения, запущенного, например с помощью uvicorn нужно следующее:

async def app(scope, receive, send):
    """
    Echo the request body back in an HTTP response.
    """
    body = await read_body(receive)
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ]
    })
    await send({
        'type': 'http.response.body',
        'body': body,
    })

То есть приложение должно принять три параметра - scope, receive, send. Где scope это данные о запроса, включая путь, receive это тело запроса, а send метод для отправки ответа.

Но нам не нужно в статье, рассматривающей FastApi писать это с нуля, так как приемом запроса и отправкой ответа занимается starlette, нам, что и делает FastApi, нужно лишь использовать соответствующие классы.

Так вот, в нашем коде, в методе инициализации класса ApiRoute функция request_response, которая, из названия ясно, используется одновременно и для обработки request, и для отправки response, принимает пользовательскую функцию, которая принимает обработанный request с помощью класса Request от starlette, и возвращает функцию или класс response, который при вызове __call__ принимает три ASGI параметра, описанных выше, и использует await send для отправки ответа, что у нас благодаря starlette уже есть, а именно класс Response.

Соответственно, наша функция get_request_handler:

self.app = request_response(get_request_handler(dependant=self.dependant))

должна быть такой, как описано выше.

Давайте взглянем на нее.

myfastapi::routing::get_request_handler

def get_request_handler(
    dependant: Dependant,
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
    is_coroutine = asyncio.iscoroutinefunction(dependant.call)
    async def app(request: Request) -> Response:
	  body = None
        if dependant.body_params:
            body = await request.json()
            буду
      solved_result = await solve_dependencies(
            request=request,
            dependant=dependant,
            body=body
            )
        values, errors = solved_result
        if errors: raise ValidationError(errors, RequestErrorModel)

        raw_response = await run_endpoint_function(
                dependant=dependant, values=values, is_coroutine=is_coroutine
            )
        if isinstance(raw_response, Response): return raw_response
	  if isinstance(raw_response, (dict, str, int, float, type(None))):
            return JSONResponse(raw_response)
        else: raise Exception("Type of response is not supported yet.")

    return app

Итак, get_request_handler принимает созданный нами ранее dependant для роута, после чего проверяет на асинхронность endpoint для роута, который хранится в переменной call у dependant.

Функция app будет использоваться во время запроса к роуту, в котором она находится. Функциональность данной функции во время запроса выглядит следующим образом. Сначала она получает тело запроса с помощью json функции класса Request от starlette, после чего, вызывает функцию solve_dependencies, которая производит валидацию полученных параметров, и возвращает необходимые значения для сигнатуры endpoint и ошибки, если они были получены при валидации. Если ошибок нет, то вызывается наша оригинальная функция run_endpoint, которая используя полученные значения параметров из запроса вызывает endpoint, который и возвращает response. Данный response как уже было описано выше, должен принимать ASGI параметры и отправлять ответ. И данный ответ, легко можно сделать, просто обернув наш ответ, например JsonResponse от starlette.

Теперь давайте рассмотрим solve_dependencies, а именно то, как она производит валидацию параметров.

myfastapi::dependencies::utils::solve_dependencies

async def solve_dependencies(
    *,
    request: Request,
    dependant: Dependant,
    body: Dict[str, Any],
) -> Tuple[Dict[str, Any], List[ErrorWrapper], Response]:
    values: Dict[str, any] = {}
    errors: List[ErrorWrapper] = []
  
    path_values, path_errors = request_params_to_args(
        dependant.path_params, request.path_params
    )
    query_values, query_errors = request_params_to_args(
        dependant.query_params, request.query_params
    )
    values.update(path_values)
    values.update(query_values)
    errors += path_errors + query_errors

    if dependant.body_params:
        (
            body_values,
            body_errors,
        ) = await request_body_to_args(  # body_params checked above
            required_params=dependant.body_params, received_body=body
        )
        values.update(body_values)
        errors.extend(body_errors)
    return values, errors

Вот наконец-то мы видим где используется dependant. Используя две довольно похожие функции request_params_to_args и request_body_to_args, которые принимают ожидаемые параметры, которые хранятся в списках класса Dependant в виде экземпляров класса ModelField от pydantic, магию которого мы увидим в этих методах, и полученные параметры, которые любезно достаются из scope классом Request от starlette, и body, который мы уже достали в get_request_handler.

Посмотрим сначала на request_params_to_args

myfastapi::dependencies::utils::request_params_to_args

from pydantic.error_wrappers import ErrorWrapper
from pydantic.errors import MissingError
 

def request_params_to_args(
    required_params: Sequence[ModelField],
    recieved_params: Union[Mapping[str, Any], QueryParams]
) -> Tuple[Dict[str, Any], List[ErrorWrapper]]:
    values: Dict[str, Any] = {}
    errors: List[ErrorWrapper] = []
    for field in required_params:
        value = recieved_params.get(field.alias)
        if value is None:
            if field.required:
                errors.append(ErrorWrapper(
                    MissingError(),
                    loc=field.alias)
                    )
            else: values[field.name] = deepcopy(field.default)
            continue

        v_, errors_ = field.validate(value, values, loc=field.alias)

        if isinstance(errors_, ErrorWrapper):
            errors.append(errors_)
        elif isinstance(errors_, list):
            errors.extend(errors_)
        else:
            values[field.name] = v_
    return values, errors

Здесь происходит следующее: мы проходим циклом по каждому параметру, который нам требуется для endpoint, после чего используя переменную alias, значение которой, это имя параметра, мы пытаемся получить значение для данного параметра из объекта полученных параметров для запроса. Если значения для параметра нету и он является обязательным, то список ошибок добавляется MissingError, если он необязательный и значения нет, то просто копируется дефолтное значение и цикл переходит к следующему параметру.

Если же значение есть, то, храня параметр со всеми его данными в виде экземпляра класса ModelField от pydantic, нам всего лишь нужно вызвать метод validate с полученным значениями, которые вернет значение для параметра и ошибки если они возникли при валидации.

Соответственно, используя request_params_to_args, мы проверяем параметры пути и запроса:

    path_values, path_errors = request_params_to_args(
        dependant.path_params, request.path_params
    )
    query_values, query_errors = request_params_to_args(
        dependant.query_params, request.query_params
    )
    values.update(path_values)
    values.update(query_values)
    errors += path_errors + query_errors

Теперь посмотрим на похожий метод request_body_to_args.

myfastapi::dependencies::utils::request_body_to_args

async def request_body_to_args(
    required_params: List[ModelField],
    received_body: Union[Dict[str, any], None],
) -> Tuple[Dict[str, any], List[ErrorWrapper]]:
    values: Dict[str, Any] = {}
    errors: List[ErrorWrapper] = []
    if required_params:
        field_alias_omitted = len(required_params)
        if field_alias_omitted == 1:
            field = required_params[0]
            received_body = {field.alias: received_body}
      
        for field in required_params:
            if field_alias_omitted:
                loc = ("body",)
            else:
                loc = ("body", field.alias)
            value: Optional[Any] = None
            if received_body is not None:
                try: value = received_body.get(field.alias)
                except AttributeError:
                    errors.append(ErrorWrapper(MissingError(), loc=loc))
                    continue
            if value is None:
                if field.required: errors.append(ErrorWrapper(MissingError(), loc=loc))
                else: values[field.name] = deepcopy(field.default)
                continue

            v_, errors_ = field.validate(value, values, loc=loc)

            if isinstance(errors_, ErrorWrapper):
                errors.append(errors_)
            elif isinstance(errors_, list):
                errors.extend(errors_)
            else:
                values[field.name] = v_
    return values, errors

Как мы видим, здесь практически все тоже самое как и в request_params_to_args, только теперь мы проверяем количество необходимых параметров в теле запроса, и идем циклам по ним.

Соответственно, мы используем функцию request_body_to_args, если список body_params класса Dependant, который хранит необходимые параметры для тела запроса, выявленные с помощью инспектирования сигнатуры endpoint, является не пустым.

if dependant.body_params:
        (
            body_values,
            body_errors,
        ) = await request_body_to_args(  # body_params checked above
            required_params=dependant.body_params, received_body=body
        )
        values.update(body_values)
        errors.extend(body_errors)
    return values, errors

Полученные значения и ошибки возвращаются в функцию get_request_handler для дальнейшего использования:

  solved_result = await solve_dependencies(
              request=request,
              dependant=dependant,
              body=body
              )
  values, errors = solved_result
  if errors: raise ValidationError(errors, RequestErrorModel)
  
  raw_response = await run_endpoint_function(
          dependant=dependant, values=values, is_coroutine=is_coroutine
      )

Теперь, имея полученные данные, которые прошли валидацию, мы можем запустить endpoint роута используя функцию run_endpoint_function:

async def run_endpoint_function(
    *, dependant: Dependant, values: List[str, Any], is_coroutine: bool
) -> Any:
    assert dependant.call is not None, "dependant.call must be a function"

    if is_coroutine:
        return await dependant.call(**values)
    else:
        return await run_in_threadpool(dependant.call, **values)

В ней мы проверяем, если endpoint является асинхронным, то запускаем его в обычном await, если же нет, мы используем функцию run_in_threadpool от starlette, чтобы запустить endpoint в асинхронном режиме, не блокируя event loop.

И сейчас, после того, как наш endpoint отработал и вернул ответ, нам нужно проверить, что это за ответ:

if isinstance(raw_response, Response): return raw_response
if isinstance(raw_response, (dict, str, int, float, type(None))):
    return JSONResponse(raw_response)
else: raise Exception("Type of response is not supported yet.")

Что здесь происходит? Сперва мы проверяем, является ли ответ экземпляром класса Response от starlette, который имеет всю необходимую функциональность для отправки ответа ASGI приложению.

Для этого нужно импортировать класс JSONResponse:

from starlette.responses import JSONResponse

и использовать его для создания ответа в endpoint

@app.post("/items/{item_id}")
async def create_item(item_id: int):
    return JSONResponse({"item": item_id})

У нас, в случае если ответ, это dict или простой тип, то мы просто оборачиваем их в JSONResponse, но у FastApi идет более сложная обработка, так, что нужно понять, это, используя класс JSONResponse в вашем пользовательском endpoint вы увеличиваете скорость ответа вашего приложения, что может быть очень важно. Также, если вы снова обратите внимание на run_endpoint_function, то поймете, что сразу делая endpoint асинхронным, вы также сокращаете время обработки.

Вот так вот, понимая внутреннее устройство, можно увеличить производительность вашего FastApi приложения - используйте асинхронность и Response от starlette для создания ваших endpoint’s.

Собственно вот, мы рассмотрели, как сделать простое FastApi приложение с нуля. Но как вы поняли, это совсем не с нуля. FastApi построен не на pydantic, как это указано в документации, а полностью на starlette.

Запускаем наше приложение

Все, что нам нужно сделать, это в первом куске кода в начале статьи заменить импорт главного класса, из которого создается приложение:

from typing import Union

from myfastapi.applications import FastAPI
from starlette.responses import JSONResponse
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None


app = FastAPI()


@app.post("/items/{item_id}")
async def create_item(item: Item, item_id: int):
    return JSONResponse({"item": item.dict(), "item_id": item_id})

@app.get("/")
async def homepage():
    return JSONResponse({'hello': 'world'})

@app.get("/get_items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

Просто используйте следующую команду:

uvicorn main:app --reload

Теперь вы можете протестировать эти API's с помощью, например, Postman.

Исходный код

Удачи!