Пособие по работе с FastAPI

Введение

FastAPI — это современный, быстрый (высокопроизводительный) веб-фреймворк для создания API на языке Python 3.7+ на основе стандартных подсказок типов Python. Он сочетает в себе лучшие практики и возможности нескольких популярных фреймворков, таких как Flask и Django, и добавляет к ним современные возможности, такие как асинхронность и автоматическую генерацию документации.

Преимущества FastAPI:

Высокая производительность: FastAPI является одним из самых быстрых фреймворков для создания API на Python.
Простота использования: Интуитивно понятный синтаксис и автоматическая генерация документации делают FastAPI доступным для разработчиков любого уровня.
Асинхронность: Поддержка асинхронных запросов позволяет обрабатывать несколько запросов одновременно, что повышает производительность.
Автоматическая документация: FastAPI автоматически генерирует интерактивную документацию с помощью Swagger UI и ReDoc.
Поддержка стандартных подсказок типов: Использование стандартных подсказок типов Python делает код более читаемым и поддерживаемым.

Установка и настройка

Установка FastAPI

Для установки FastAPI выполните следующую команду:

pip install fastapi uvicorn

Где:

fastapi — это сам фреймворк.
uvicorn — это ASGI-сервер, который используется для запуска приложений FastAPI.

Создание первого приложения

Создайте файл main.py и добавьте в него следующий код:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

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

Для запуска приложения выполните следующую команду:

uvicorn main:app --reload

Где:

main — это имя файла (без расширения .py).
app — это имя переменной, в которой хранится экземпляр FastAPI.
--reload — это флаг, который включает автоматическую перезагрузку сервера при изменении кода.

После запуска приложения вы можете открыть браузер и перейти по адресу http://127.0.0.1:8000, чтобы увидеть автоматически сгенерированную документацию Swagger UI.

Основные концепции

Маршруты

Маршруты в FastAPI определяются с помощью декораторов, таких как @app.get, @app.post, @app.put, @app.delete и других. Эти декораторы указывают, какой HTTP-метод должен использоваться для доступа к определенному маршруту.

Пример определения маршрутов:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.post("/items/")
def create_item(item: dict):
    return {"item": item}

@app.put("/items/{item_id}")
def update_item(item_id: int, item: dict):
    return {"item_id": item_id, "item": item}

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    return {"item_id": item_id}

Запросы и ответы

FastAPI предоставляет простые и интуитивно понятные способы обработки запросов и формирования ответов. Вы можете использовать стандартные типы данных Python для определения структуры запросов и ответов.

Пример обработки запросов и формирования ответов:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

Параметры маршрутов

FastAPI поддерживает параметры маршрутов, которые позволяют извлекать данные из URL. Вы можете определить параметры маршрутов, добавив их в путь маршрута в фигурных скобках.

Пример использования параметров маршрутов:

from fastapi import FastAPI

app = FastAPI()

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

@app.get("/users/{user_id}")
def read_user(user_id: str):
    return {"user_id": user_id}

Параметры запросов

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

Пример использования параметров запросов:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

Работа с базой данных

FastAPI легко интегрируется с различными базами данных, включая SQLite, PostgreSQL, MySQL и другими. Вы можете использовать ORM (Object-Relational Mapping) библиотеки, такие как SQLAlchemy или Tortoise-ORM, для работы с базами данных.

Пример работы с SQLite

Установите SQLAlchemy и SQLite:

pip install sqlalchemy databases

Создайте файл database.py и добавьте в него следующий код:

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)
    price = Column(Integer)

Base.metadata.create_all(bind=engine)

Создайте файл main.py и добавьте в него следующий код:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import database, models, schemas

app = FastAPI()

# Dependency
def get_db():
    db = database.SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/items/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

Асинхронность в FastAPI

FastAPI поддерживает асинхронные запросы, что позволяет обрабатывать несколько запросов одновременно. Это особенно полезно для операций ввода-вывода, таких как запросы к базе данных или внешним API.

Пример асинхронного маршрута

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/async")
async def read_async():
    await asyncio.sleep(1)
    return {"message": "Async response"}

Пример асинхронной работы с базой данных

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from . import database, models, schemas

app = FastAPI()

# Dependency
async def get_db():
    async with database.SessionLocal() as db:
        yield db

@app.get("/items/{item_id}", response_model=schemas.Item)
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(models.Item).where(models.Item.id == item_id))
    db_item = result.scalars().first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

Использование мидлваров

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

Пример мидлвара для логирования

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Пример мидлвара для аутентификации

from fastapi import FastAPI, Request, HTTPException
from fastapi.security import HTTPBearer

app = FastAPI()
security = HTTPBearer()

@app.middleware("http")
async def authenticate(request: Request, call_next):
    if request.url.path.startswith("/protected"):
        auth = request.headers.get("Authorization")
        if not auth or not auth.startswith("Bearer "):
            raise HTTPException(status_code=401, detail="Unauthorized")
    response = await call_next(request)
    return response

Документация с помощью Swagger

FastAPI автоматически генерирует интерактивную документацию с помощью Swagger UI и ReDoc. Вы можете получить доступ к документации, перейдя по адресу http://127.0.0.1:8000/docs для Swagger UI или http://127.0.0.1:8000/redoc для ReDoc.

Настройка документации

Вы можете настроить документацию, добавив дополнительную информацию о вашем API:

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="This is a very fancy API",
    version="1.0.0",
    contact={
        "name": "API Support",
        "url": "http://example.com/support",
        "email": "support@example.com",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)

Пример проекта

Структура проекта

my_fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   ├── schemas.py
│   └── crud.py
├── requirements.txt
└── README.md

Файл app/main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import database, models, schemas, crud

app = FastAPI()

# Dependency
def get_db():
    db = database.SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/items/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    return crud.create_item(db=db, item=item)

@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = crud.get_item(db=db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db=db, skip=skip, limit=limit)
    return items

Файл app/database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Файл app/models.py

from sqlalchemy import Column, Integer, String
from .database import Base

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)
    price = Column(Integer)

Файл app/schemas.py

from pydantic import BaseModel

class ItemBase(BaseModel):
    name: str
    description: str = None
    price: int

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int

    class Config:
        orm_mode = True

Файл app/crud.py

from sqlalchemy.orm import Session
from . import models, schemas

def get_item(db: Session, item_id: int):
    return db.query(models.Item).filter(models.Item.id == item_id).first()

def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()

def create_item(db: Session, item: schemas.ItemCreate):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

Файл requirements.txt

fastapi
uvicorn
sqlalchemy
databases

Файл README.md

# My FastAPI Project

This is a sample FastAPI project.

## Installation

1. Clone the repository:

```bash
git clone https://github.com/yourusername/my_fastapi_project.git

2. Install the dependencies:

pip install -r requirements.txt

3. Run the application:

uvicorn app.main:app --reload

4. Open your browser and go to http://127.0.0.1:8000/docs to see the Swagger UI.

Лучшие практики

Структурирование проекта

Разделяйте ваш проект на модули и пакеты для лучшей организации кода.
Используйте отдельные файлы для моделей, схем, маршрутов и других компонентов.
Следуйте принципам SOLID для создания поддерживаемого и расширяемого кода.

Документация

Всегда документируйте ваши API с помощью docstrings и аннотаций типов.
Используйте Swagger UI и ReDoc для автоматической генерации документации.
Добавляйте примеры запросов и ответов для лучшего понимания вашего API.

Тестирование

Пишите тесты для ваших маршрутов и функций.
Используйте библиотеки для тестирования, такие как pytest и httpx.
Тестируйте как синхронные, так и асинхронные маршруты.

Безопасность

Используйте мидлвары для аутентификации и авторизации.
Защищайте ваши маршруты от атак, таких как CSRF и XSS.
Используйте HTTPS для защиты данных, передаваемых по сети.

Производительность

Используйте асинхронные маршруты для операций ввода-вывода.
Оптимизируйте запросы к базе данных для уменьшения времени ответа.
Используйте кэширование для уменьшения нагрузки на сервер.

Заключение

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

Полезные ресурсы

Если у вас есть дополнительные вопросы или требуется помощь, дайте знать!

ИИ-помощник

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

Привет! Я помогу разобраться с материалом урока. Просто напишите вопрос.