Если вы сейчас разрабатываете новое приложение на Python, высока вероятность, что при этом вы используете FastAPI. В FastAPI заложено множество отличных возможностей, благодаря которым с ним легко начинать работу. Но в FastAPI есть и немало нюансов, на понимание которых требуется время. Мне пришлось особенно попотеть с одним аспектом, а именно — как FastAPI управляет вызовами к маршрутам API через декорированные параметры пути. Давайте подробно об этом поговорим.

Что происходит на веб-сервере

Одним из важнейших компонентов любого веб-приложения (которое мы создаём) является веб-сервер, программа, слушающая входящие запросы, поступающие из сети. Затем она транслирует эти запросы в методы, которые, в свою очередь, вызываются на бэкенде.

Чтобы лучше понимать, что здесь происходит под капотом, давайте сначала реализуем простой веб-сервер. Для этого воспользуемся модулем http.server, который входит в стандартную библиотеку Python.

Нам требуется написать программу, которая слушает порт и принимает HTTP-запросы. А именно: принимает запрос, разбирает маршрут пути, а также разбирает любые данные, прикреплённые к HTTP-вызову. См. также “All I want is to cURL and parse a JSON object”.

import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

class RequestHandler(BaseHTTPRequestHandler):

    def parse_path(self, request_path: str)-> dict:
        """
        Parse request path
        """
        parsed = urlparse(request_path)
        print(parsed)
        params_dict = parse_qs(parsed.query)
        return params_dict

    def store_urls(self, request_path: str)-> None:
        """
        Parse URLs and store them
        """
        params = self.parse_path(request_path)
        print(params)
        for key, val in params.items():
            self.data_store.put_data(val[0])


    def return_k_json(self, k:dict)-> BinaryIO:
        """
        Return json response
        """
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        # Содножит поток вывода для записи отклика обратно на клиент. 
        # BufferedIOBase пишет в поток
        # См. https://docs.python.org/3/library/io.html#io.BufferedIOBase.write
        self.wfile.write(json.dumps(k).encode('utf-8'))


    def bad_request(self):
        """
        Handle bad request
        """
        self.send_response(400)
        self.send_header("Content-type", "application/json")
        self.end_headers()

    def do_GET(self):
         
        request_path = self.path

        if self.path == "/":
            self.return_k_json({"ciao": "mondo"})
        if request_path.startswith("/get"):
            key = self.parse_path(request_path)
            self.return_k_json({"jars": key["key"]})
            self.send_response(200)
        else:
            self.bad_request()
            self.end_headers()

    def do_POST(self):
        request_path = self.path

        if request_path.startswith("/set"):
            self.store_urls(request_path)
            self.send_response(200)
        else:
            self.bad_request()


if __name__ == "__main__":
    host = "localhost"
    port = 8000

    server = HTTPServer((host, port), RequestHandler)
    print("Server started http://%s:%s" % (host, port))
    server.serve_forever()

Что тут происходит?

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

Для начала спроектируем простейший API. Вот что мы хотим сделать как его пользователи:

• Протестировать сервис и получить в ответ простой отклик
• Добавлять новые баночки в список имеющихся
• Просматривать добавленные нами баночки

Преобразуем эти действия в запросы GET и PUT, так, чтобы для них можно было написать HTTP-вызовы. Чтобы не усложнять пример, мы даже не будем хранить их на стороне сервера, а запишем их. Так мы без труда сможем просмотреть, как отправлять данные в наше приложение.

Давайте протестируем сервер:

> python serve.py
> curl -X POST http://localhost:8000/
> {"ciao": "mondo"}

Мы хотим сохранять элементы:

> curl -X POST http://localhost:8000/set\?key\=8
200 OK

И получать сохранённые элементы обратно:

> curl -X GET http://localhost:8000/get\?key\=8 
> {"jars": ["8"]}

Нужно предусмотреть на нашем сервере механизм, который обеспечивал бы синтаксический разбор всех получаемых информационных фрагментов:

1. Тип запроса. В реализации HTTP do_GET и do_POST неявно обрабатывают это.
2. Параметры, передаваемые нами в запрос пути, так, чтобы с ними можно было что-нибудь проделать
3. Маршрут к тому методу внутри нашего приложения, который и обрабатывает данные

В нашем просто сервере вся суть маршрутизации происходит на уровне методов. Если отправить базовый путь, то программа вернёт {«ciao»: «mondo»}. В противном случае будет возвращена информация о том, сколько баночек мы передали через путь запроса (эту информацию получаем в результате разбора параметров пути).

def do_GET(self) -> None:
    request_path = self.path

    if self.path == "/":
        self.return_k_json({"ciao": "mondo"})
    if request_path.startswith("/get"):
        key = self.parse_path(request_path)
        # Здесь идёт действие, выполняемое внутри веб-приложения 
        self.return_k_json({"jars": key["key"]})
        self.send_response(200)
    else:
        self.bad_request()
        self.end_headers()

Как видите, ситуация быстро усложняется. Например, что будет, если мы станем выполнять множество операций в рамках запроса GET. Допустим, станем вытягивать информацию из базы данных, или из кэша, или извлекать ресурсы? У нас будут различные методы, которые мы станем обрабатывать в зависимости от того, по какому пути пошёл разбор. Что, если у нас также будут операторы PUT/DELETE? Что, если нам потребуется аутентификация? Запись в базу данных? Статические страницы? Код постоянно усложняется относительно исходного состояния, и на данном этапе нам уже требуется фреймворк.

Starlette

Среди первых фреймворков Python для веб-разработки были могучие Django и Flask. В новейшее время, когда в Python стали усиливаться позиции асинхронной обработки, на сцену вышли такие фреймворки как Starlette, прямо из коробки предоставляющие функционал асинхронной обработки.

Автор Starlette также создал Django Rest Framework. В Starlette включены легковесные операции, обеспечивающие базовый функционал HTTP-вызовов, а наряду с ним — например, работу с веб-сокетами. В качестве бонуса все эти операции по умолчанию выполняются асинхронно.

Чтобы управлять HTTP-вызовами по тому же принципу, как и с нашим простым веб-сервером, в Starlette можно сделать следующее:

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def homepage(request):
    return JSONResponse({'ciao': 'mondo'})


app = Starlette(debug=True, routes=[Route('/', homepage),])

Мы запускаем экземпляр приложения Starlette, в котором есть маршруты процессов. На уровне пути каждый маршрут связан именно с тем методом, который он вызывает. Если Starlette видит данный конкретный маршрут, то вызывает нужный метод с учётом логики синтаксического разбора и чтения заголовков и тел HTTP-запросов.

А что, если добавить второй вызов метода, привязанный уже к другому маршруту и также возвращающий нам количество баночек?

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({'ciao': 'mondo'})

async def get_jars(request):
    return JSONResponse({'jars': ['8']})

app = Starlette(debug=True, routes=[
    Route('/', homepage),
    Route('/get_jars', get_jars)
])

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

Реализация FastAPI

FastAPI обёртывает Starlette – поскольку, как сказано в документации, «фактически, представляет собой Starlette на стероидах» — и включает валидацию типов по модели Pydantic на логических границах приложения.

Когда мы создаём экземпляр приложения FastAPI, под капотом он остаётся «просто» экземпляром приложения Starlette, свойства которого мы переопределяем на уровне приложения.

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"ciao": "mondo"}

@app.get("/jars/{id}")
async def get_jars(id):
    return {"message": f"jars: {id}"}

При разработке FastAPI использует uvicorn. Это ASGI-сервер, при помощи которого он слушает входящие запросы и обрабатывает их в соответствии с маршрутами, определёнными у вас в приложении.

Uvicorn инициализирует ASGI-сервер, связывает его с сокетными соединениями на порте 8000 и начинает слушать входящие соединения. Поэтому, когда мы отправляем запрос GET по главному маршруту, который по умолчанию располагается на порте 8000, мы рассчитываем получить в ответ ciao mondo.

FastAPI, как и в случае с предыдущими приложениями, по-прежнему делегирует операции с путями и соответствующие методы маршрутизатору, обрабатывающему их и разбирающему параметры, но обёртывает всё это в декоратор Python. Такой код проще написать, но сложнее понимать в том отношении, как именно происходит обработка пути.

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

Без простого сервера мы:

1. Запускаем сервер
2. Слушаем входящие запросы на порте 8000
3. Получив запрос, направляем его методу do_GET
4. В зависимости от пути запроса, направляем его в "/"
5. Возвращаем запрос клиенту с кодом состояния 200

В FastAPI мы:

1. Запускаем веб-сервер uvicorn (если находимся в режиме разработки, в производстве же придётся выбрать совместимый рабочий класс)
2. Слушаем входящие запросы на порте 8000
3. Создаём экземпляр приложения FastAPI
4. Он, в свою очередь, создаёт экземпляр Starlette
5. Получив запрос GET, направляем его в метод self.get нашего приложения
6. Он, в свою очередь, вызывает self.router.get с применением операции пути
7. Маршрутизатор — это экземпляр routing.APIRouter
8. Метод .get в APIRouter принимает путь и возвращает return self.api_route. Именно в этой точке фактически вызывается декоратор – мы видим, как декоратор в этом методе принимает в качестве ввода функцию DecoratedCallable и возвращает декорированный add_api_route, который и добавляет этот маршрут в список имеющихся.

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

Маршрутизация параметров пути

Маршрутизация параметров пути происходит в Starlette, где происходит синтаксический разбор параметров пути, относящихся к запросу, и они записываются в словарь (точно как и в случае с простым веб-приложением). Эта задача решается при помощи шаблонизатора Jinja.

TL;DR

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

P.S. Обращаем ваше внимание на то, что у нас на сайте проходит распродажа.