Введение FastAPI

Если вы разрабатываете свои проекты на Django , то вы легко можете разработать API с помощью замечательной библиотеки Django REST framework(DRF). Но иногда бывает полезно рассматривать альтернативы и другие технологии , которые специально заточены для реализации данных задач. И при том эти фреймворки учитывают плюсы и минусы существующих технологий и разрабатываются с учетом новых возможностей языка Python.  Одним из таких современных фреймворков , который разрабатывался для построения API является фреймворк FastAPI.

Многие в сети советуют фреймворк Flask для написания API , но использовать еще один синхронный фреймворк для меня является нецелесообразным , учитывая , что есть такой прекрасный асинхронный фреймворк FastAPI, который базируется на другом замечательном асинхронном фреймворке Starlette. Если вы еще не писали асинхронный код и не использовали новые возможности питона , такие как подсказки типов(type hinting) , то вам нужно использовать FastAPI.

Из требований вы должны использовать Python 3.6 и выше.

Мы создадим API для наших постов. Реализуем все методы CRUD (Create, Read, Update, Delete)

Устанавливаем FastAPI

pip install fastapi

В качестве зависимостей данный фреймворк тянет за собой фреймворк Starlette и библиотеку Pydantic, которая используется для парсинга и валидации данных

Для запуска приложений на FastAPI нам нужен ASGI сервер. Поэтому мы установим Uvicorn

pip install uvicorn[standard]

Для хранения наших постов мы будем использовать базу данных SQLite. В Django для работы с базой данных у нас есть встроенный Django ORM. Обычно для работы с БД на других фреймворках Python я использую SQLAlchemy , которая в своем составе имеет SQLAlchemy ORM. Но так как на FastAPI мы с БД ходить взаимодействовать в асинхронном режиме , то мы не можем использовать SQLAlchemy ORM. Поэтому будем использовать SQLAlchemy Core , который поддерживает асинхронный стиль.

Поэтому мы установим encode/databases для SQLite следующим образом

pip install databases[sqlite]

Тут мы используем SQLite для нашего примера (так как его легко использовать), а так на продакшене я бы рекомендовал использовать PostgreSQL. Для PostgreSQL мы encode/databases должны установить следующим образом

pip install databases[postgresql]

Для описания таблицы posts мы должны импортировать библиотеку sqlalchemy , создать metadata объект и с помощью sqlalchemy.Table укажем , какие столбцы будет иметь наша таблица .

import sqlalchemy
metadata = sqlalchemy.MetaData()

posts = sqlalchemy.Table(
    "posts",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.INTEGER, primary_key=True),
    sqlalchemy.Column("title", sqlalchemy.String),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("is_published", sqlalchemy.Boolean),
)

Далее подключимся к SQLite и вызовем metadata.create_all , которая создаст таблицу в БД , если ее там нет.

import databases
DATABASE_URL = "sqlite:///./blogapi.db"
database = databases.Database(DATABASE_URL)
engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

metadata.create_all(engine)

Создадим Pydantic модели. Это позволит нам валидировать входные данные и если входящие данные не будут соответствовать типам , то возникнут исключения валидации. Также входящие данные будут сериализированы и документированы и это мы можем увидеть в документации OpenAPI , которая генерируется автоматически. Документация доступна в двух форматах: Swagger и Redoc.

from pydantic import BaseModel


class PostIn(BaseModel):
    title: str
    text: str
    is_published: bool


class Post(BaseModel):
    id: int
    title: str
    text: str
    is_published: bool

Ниже я покажу как использовать эти модели.

Сейчас нам необходимо создать приложение на FastAPI , а также создадим обработчики событий для подключения и для разъединения к БД.

app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

Полный листинг кода файла main.py. Также добавил исходный код на Github.

from typing import List
import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel


DATABASE_URL = "sqlite:///./blogapi.db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

posts = sqlalchemy.Table(
    "posts",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.INTEGER, primary_key=True),
    sqlalchemy.Column("title", sqlalchemy.String),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("is_published", sqlalchemy.Boolean),
)

engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

metadata.create_all(engine)

class PostIn(BaseModel):
    title: str
    text: str
    is_published: bool


class Post(BaseModel):
    id: int
    title: str
    text: str
    is_published: bool

app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get('/posts/', response_model=List[Post])
async def read_posts():
    query = posts.select()
    return await database.fetch_all(query)


@app.post("/posts/", response_model=Post)
async def create_post(post: PostIn):
    query = posts.insert().values(title=post.title, text=post.text, is_published=post.is_published)
    last_record_id = await database.execute(query)
    return {**post.dict(), "id": last_record_id}


@app.delete("/posts/{post_id}")
async def delete_post(post_id: int):
    query = posts.delete().where(id == post_id)
    await database.execute(query)
    return {"detail": "Post deleted", "status_code": 204}

Через командную строку запустим наш проект на FastApi используя ASGI-сервер uvicorn.

uvicorn main:app

Большой плюс FastAPI в том , что он автоматически генерирует документацию OpenAPI, которая доступна в формате Swagger по адресу: http://127.0.0.1:8000/docs

Документация OpenAPI в формате Redoc доступна по адресу: http://127.0.0.1:8000/redoc

Заключение

В данной статье мы написали простой API для постов ,с помощью фреймворка FastAPI