Как запустить Fastapi в Yandex Cloud Serverless Containers
В принципе весь процесс описан в документации, но там упомянут всего один фреймворк - Sanic, а тут мы разберем как запустить Fastapi.
Минимальный проект
Создаем проект
Для этого воспользуемся еще одной инструкцией, на этот раз из документации Fastapi
mkdir app
touch app/__init__.py
touch app/main.py
git init
cat << EOF > requirements.txt
fastapi
pydantic
uvicorn
EOF
git add .
git commit -m "Initial commit"
Наполняем проект
В app/app.py добавим простой хэндлер.
import os
from typing import Union
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: Union[str, None] = None):
return {"item_id": item_id, "q": q}
Дополнительно к примеру из документации Fastapi,
мы добавим main.py, чтобы программно запустить сервер через вызов uvicorn.run().
В параметрах запуска сервера мы указываем host и port, который берем из переменных окружения.
Это требование описано в облачной документации.
import os
import uvicorn
from app.app import app
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")), log_config=None)
Создаем Dockerfile
# Для основы возьмем образ с Python 3.9
FROM python:3.9
# Установим рабочую директорию
WORKDIR /code
# Скопируем файлы с зависимостями
COPY ./requirements.txt /code/requirements.txt
# Установим зависимости
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
# Скопируем файлы проекта
COPY ./main.py /code
COPY ./app /code/app
# Запустим сервер
CMD python main.py
Собираем контейнер
Для этого подготовим Makefile
REVISION := $(shell git rev-parse --short HEAD)
DOCKER_IMAGE_NAME = cr.yandex/your_cr_id/fatsapi:$(REVISION)
build:
docker build --platform linux/amd64 -t $(DOCKER_IMAGE_NAME) .
push:
docker push $(DOCKER_IMAGE_NAME)
run:
docker run --rm -e PORT=8080 -p 8080:8080 $(DOCKER_IMAGE_NAME)
Теперь мы можем собрать и запушить контейнер
make build
make push
Отлично, теперь мы можем запустить контейнер локально
make run
Запускаем в Yandex Cloud Serverless Containers
Для создания ревизии необходимо указать имя и тег контейнера, а также указать сервисный аккау нт, который будет
использоваться для запуска контейнера. У него должны быть права на запуск пулл образов из Container Registry.
Например, роль container-registry.images.puller на каталог или реестр, в которых находится Docker-образ.
Если все прошло успешно, то при переходе по публичной ссылке из консоли, мы должны увидеть следующий результат:
{"Hello": "World"}
Отлично. Базовый сценарий работает. Давайте теперь добавим улучшений, которые помогут удобнее анализировать работу контейнера.
Добавляем структурированные логи
Если мы запускаем наш текущий пример, то мы увидим в логах следующее:
INFO: Started server process [7]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
INFO: 172.17.0.1:48202 - "GET / HTTP/1.1" 200 OK
В сервисе просмотра логов в облаке это будет выглядеть так:
Для начала не плохо, но мы можем сделать лучше. Для этого воспользуемся структурированными логами.
Добавляем логгер
import json
import logging
from logging import Formatter
class JsonFormatter(Formatter):
def __init__(self):
super(JsonFormatter, self).__init__()
def format(self, record):
json_record = {}
json_record["message"] = record.getMessage()
json_record["level"] = str.replace(str.replace(record.levelname, "WARNING", "WARN"), "CRITICAL", "FATAL")
return json.dumps(json_record)
logger = logging.root
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.handlers = [handler]
logger.setLevel(logging.DEBUG)
Дополним наш main.py следующим образом:
import uvicorn
from fastapi import FastAPI
+ from src.logger import logger
app = FastAPI()
@app.get("/")
async def hello():
+ logger.info("Hello")
return "Hello World"
if __name__ == "__main__":
- uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")), log_level="info")
+ uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")), log_config=None)
Теперь мы можем запустить контейнер локально и увидеть в логах следующее:
{"message": "Hello", "level": "INFO"}
Неплохое начало. Но можно сделать лучше.
Добавляем контекст
Если в рамках одного запроса мы будем писать в логи, то нам будет полезно знать, какой запрос сейчас обрабатывается. Для того чтобы собрать все записи, мы можем писать в логи идентификатор запроса. Для этого нам необходимо добавить middleware, который будет добавлять в контекст запроса идентификатор.
pip install asgi_correlation_id
Теперь если пользователь в запросе передаст заголовок X-Request-ID, то мы будем использовать его в качестве
идентификатора запроса. Если заголовок не передан, то мы сгенерируем случайный идентификатор. Получить идентификатор
можно вызвав correlation_id.get().
import json
import logging
from logging import Formatter
+ from asgi_correlation_id import correlation_id
class JsonFormatter(Formatter):
def __init__(self):
super(JsonFormatter, self).__init__()
def format(self, record):
json_record = {}
json_record["message"] = record.getMessage()
+ json_record["request_id"] = correlation_id.get()
json_record["level"] = str.replace(str.replace(record.levelname, "WARNING", "WARN"), "CRITICAL", "FATAL")
if "req" in record.__dict__:
json_record["req"] = record.__dict__["req"]
if "res" in record.__dict__:
json_record["res"] = record.__dict__["res"]
if record.levelno == logging.ERROR and record.exc_info:
json_record["err"] = self.formatException(record.exc_info)
return json.dumps(json_record)
logger = logging.root
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.handlers = [handler]
logger.setLevel(logging.DEBUG)
logging.getLogger("uvicorn.access").disabled = True
+ logging.getLogger("uvicorn.asgi").disabled = True
+ logging.getLogger("uvicorn.error").disabled = True
Отключив uvicorn.asgi и uvicorn.error мы избавимся от дублирования логов, потому что переопределим функциональность их в нашем middleware.
import sys
import traceback
from time import time
from asgi_correlation_id import correlation_id
from starlette.middleware.base import BaseHTTPMiddleware
from app.logger import logger
class LogMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
try:
start = time()
response = await call_next(request)
process_time = time() - start
logger.info(
"Incoming request",
extra={
"process_time": process_time,
"req": {
"method": request.method,
"url": str(request.url),
},
"res": {"status_code": response.status_code, },
},
)
return response
except Exception:
exc_type, exc_value, exc_traceback = sys.exc_info()
logger.error(
"Internal server error",
extra={
"request_id": correlation_id.get(),
"req": {
"method": request.method,
"url": str(request.url),
},
"res": {"status_code": 500, "err": "".join(traceback.format_exception(
exc_type,
exc_value,
exc_traceback))},
},
)
Теперь в лога мы увидим записи с правильным уровнем лог ирования, по которому можно легко пофильтровать записи.
А если кликнуть на поле в логах, то мы сможем быстро добавить условие фильтрации по этому полю.
Код примера можно найти тут