Api микросервисы на FastApi и Python

Часто вижу в требованиях к DE знание фреймворка FastApi для создание конечного api к расчитанным данным.
Решил в этой статье пройтись обзорно по основным функциям.

Преимущества фреймворка

- легкий старт за счет декораторов
- встроенный веб сервер
- async io из коробки для асинхронной работы с http
- валидация данных на основе схемы
- автоматическая swagger дока
- авторизация и прочее

Docker образ с библиотеками и vscode-server

Ниже образ, который я использовал для тестов. Он включает в себя python 3.12, fastapi, request и сервер vscode для работы с образом снаружи.

FROM python:3.12-slim

SHELL ["/bin/bash", "-c"] 

RUN python3.12 -m venv ~/.api && \
    source ~/.api/bin/activate && \
    pip install fastapi[all] && \
    pip install import requests && \
    mkdir -p /opt/fastapi && \
    apt-get -y update && \
    apt-get -y install curl

# install VS Code (code-server)
RUN curl -fsSL https://code-server.dev/install.sh | sh

# install VS Code extensions
RUN code-server --install-extension redhat.vscode-yaml \
                --install-extension ms-python.python

ENTRYPOINT ["/bin/bash"]

Сборка:

docker build . -t fastapi

Маппинг папки с сервером vscode на локальную папку, чтобы каждый раз его заново не ставить:

docker run -it \
    -v /home/pihel/Documents/fastapi/vscode-server:/root/.vscode-server \
    -v /home/pihel/Documents/fastapi/code:/opt/fastapi \
    --entrypoint /bin/bash \
    --rm \
    fastapi

Первое приложение

Для быстрого старта достаточно минимума действий:

from fastapi import FastAPI

app = FastAPI()

@app.post('/calculate')
async def calculate(num1: int, num2: int):
    return {f'sum of numbers {num1} and {num2} is ': f'{num1+num2}'}

#виды параметров можно совмещать
@app.get("/users/{user_id}")
def read_user(user_id: int, is_admin: bool = False):
    return {"user_id": user_id, "is_admin": is_admin}

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

uvicorn main:app --reload

После этого приложение можно открыть по адерсу: http://127.0.0.1:8000/
Автоматически генерируется swagger дока по адресу http://127.0.0.1:8000/docs

Валидация данных

Кроме валидации происходит преобразование к датаклассу User:

from pydantic import BaseModel

class User(BaseModel):
    username: str
    message: str
    
@app.post("/")
async def root(user: User):

Передача json с данными:

{
    "username": "Vasya",
    "message": "I am BATMAN"
}

Для выходных данных тоже можно задать выходной дата тип

@app.post('/add_user', response_model=User) # тут указали модель (формат) ответа
async def add_user(user: User): # собственно тут проверяем входные данные на соответствие модели
    fake_db.append({"username": user.username, "user_info": user.user_info}) # тут добавили юзера в фейковую БД
    return user

Параметры могут передаваться в части пути, так и обычными параметрами запроса:

@app.get('/{user_id}') # тут объявили параметр пути
async def search_user_by_id(user_id: int): # тут указали его тип данных
    # какая-то логика работы поиска
    return {"вы просили найти юзера с id": user_id}

Асинхронное выполнение задачи

from fastapi import BackgroundTasks, FastAPI
app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

Куки и заголовки

Куки:

@app.get("/")
    def root(last_visit = Cookie()):
        return  {"last visit": last_visit}

Заголовки:

from fastapi import Header
@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return {"User-Agent": user_agent}

Авторизация

def authenticate_user(credentials: HTTPBasicCredentials = Depends(security)):
    user = get_user_from_db(credentials.username)
    if user is None or user.password != credentials.password:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
    return user

@app.get("/protected_resource/")
def get_protected_resource(user: User = Depends(authenticate_user)):
    return {"message": "You have access to the protected resource!", "user_info": user}

Обработка ошибок

Виды своих ошибок будут отображены в swagger доке

# не изменяли
class CustomException(HTTPException):
    def __init__(self, detail: str, status_code: int = 400):
        super().__init__(status_code=status_code, detail=detail)

# Обработчик ошибок (error handler) для класса CustomException 
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail}
    )

# не изменяли
@app.get("/items/{item_id}/")
async def read_item(item_id: int):
    if item_id == 42:
        raise CustomException(detail="Item not found", status_code=404)
    return {"item_id": item_id}

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

Тестовый клиент для обращения к апи:

from fastapi.testclient import TestClient
from my_app import app

client = TestClient(app)

def test_calculate_sum():
    response = client.get("/sum/?a=5&b=10")
    assert response.status_code == 200
    assert response.json() == {"result": 15}

Сами апи вызовы нужно мокать

Обработка событий

@app.middleware("http")
async def my_middleware(request: Request, call_next):
    print('Мидлвэр начал работу')
    response = await call_next(request)
    print('Мидлвэр получил обратно управление')
    return response

@app.get("/")
def index():
    print('привет из основного обработчика пути')
    return {"message": "Hello, world!"}

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

Скахин Алексей / pihel

Страницы

суббота, 28 декабря 2024 г.