Создавайте и размещайте приложения Fast Data Science с помощью FastAPI
Недавно я разместил ряд готовых приложений для обработки данных в виде веб-служб Restful с использованием веб-инфраструктуры FastAPI.
Я обнаружил, что FastAPI стабилен и прост в использовании, и по этим причинам я решил написать статью о библиотеке FastAPI, описывающую шаги, которые мы можем выполнить, чтобы разместить приложение для обработки данных.
FastAPI был выпущен в 2018 году и становится де-факто выбором для создания высокопроизводительных приложений для обработки данных.
Эта статья объяснит, что такое FastAPI, почему он превосходит своих конкурентов, а также пошаговое руководство о том, как разместить приложение для обработки данных с использованием FastAPI.
1. Что такое FastAPI?
Как видно из названия, FastAPI является высокопроизводительной веб-инфраструктурой. Его производительность можно сравнить с NodeJS и Go, и он считается одним из самых быстрых доступных сред Python.
Netflix, Uber, Microsoft и многие другие корпорации используют библиотеку FastAPI.
FastAPI может помочь нам создать API с Python 3.6+.
Он основан на стандартных подсказках типа Python. FastAPI не только интуитивно понятен и прост в использовании в проектах, но и код FastAPI также покрывает 100% тестов, и, следовательно, это готовый к работе надежный код.
Пользуясь Flask уже довольно давно, я заметил, что FastAPI так же прост в использовании, как Flask. Фактически, FastAPI адаптировал свою простоту от Flask API. Впоследствии, в общем, я рекомендую изучить FastAPI для будущих проектов.
FastAPI построен на библиотеках Starlette и Pydantic.
Диаграмма показывает, что FastAPI использует Starlette для своих веб-частей и Pydantic для частей данных.
Pydantic библиотека Python для выполнения проверки данных, документации по сериализации. Pydantic использует стандартные библиотеки типов typing, как это определено в PEP 484 для определения сложных объектов.
Давайте рассмотрим особенности FastAPI.
2. Основные функции FastAPI
1. Первая особенность, которую нужно выделить, - это автоматическая интерактивная документация по API
FastAPI основан на открытых стандартах для OpenAPI и JSON Schema. В результате он может автоматически создавать SwaggerUI для нашего веб-API без какого-либо дополнительного кодирования, если мы используем соответствующие типы данных Pydantic. SwaggerUI является интерактивным по своей природе и позволяет нам тестировать API непосредственно из браузера. Автоматическая документация была вдохновлена APISpec.
Я продемонстрирую, как мы можем построить документацию по автоматической модели данных с помощью схемы JSON, используя FastAPI, позже в этой статье.
FastAPI также позволяет нам проверять входные данные от внешних абонентов и генерирует автоматические ошибки для клиентов, когда он получает недействительные данные. Функция документирования была основана на DEST-среде Django.
OpenAPI автоматически генерирует схему для приложения. FastAPI также поддерживает сериализацию данных и анализ входных данных, которые были вдохновлены Marshmallow и Webargs.
2. Вторая ключевая особенность - поддержка асинхронных операций ввода-вывода.
Крайне важно создать веб-API, который не блокирует своих абонентов во время выполнения операций. Библиотека asyncio в Python предлагает большое количество асинхронных возможностей.
Библиотека FastAPI поддерживает асинхронные операции ввода-вывода. Это стало возможным благодаря тому, что FastAPI основан на библиотеке Starlette.
Сама библиотека Starlette находится на вершине Uvicorn. Uvicorn - это молниеносный сервер ASGI, поддерживающий асинхронные платформы.
Starlette - это легкая платформа ASGI, которая позволяет нам создавать высокопроизводительные асинхронные сервисы.
В отличие от сервера WSGI, серверы ASGI могут поддерживать HTTP / 2 и WebSockets. С помощью Uvicorn мы можем создать веб-API, который может обеспечить высокую пропускную способность в контексте ввода-вывода.
Starlette использует более новый сервер ASGI асинхронного ввода-вывода и также поддерживает фоновые задачи. Следовательно, мы можем создавать быстрые веб-API с малой латентностью, используя FastAPI
3. Самая важная особенность; мы можем быстро создать конечные точки REST для нашего приложения для обработки данных
Пользователи Flask должны установить пакет Flask-restplus, чтобы создать конечные точки REST для своего приложения для обработки данных.
FastAPI поддерживает Rest операции GET, PUT, POST, DELETE, OPTIONS, HEAD, PATCH и TRACE без каких-либо дополнительных пакетов. Все маршруты вместе с их операциями автоматически генерируются в документации.
4. Приложение FastAPI легко установить, использовать и поддерживать.
Недавно я заметил закономерность в отрасли. Становится жизненно важным быстро вывести решение для обработки данных на рынок. Кроме того, приложения для обработки данных являются одним из компонентов большой экосистемы, и им необходимо взаимодействовать с рядом служб и интерфейсов. Это означает, что крайне важно иметь возможность быстрого развертывания приложения в виде веб-службы, чтобы его могли использовать конечные пользователи.
FastAPI прост в использовании. Буквально за минуты требуется разместить приложение для обработки данных в качестве микросервиса и предоставить конечные точки, которые могут вызывать другие приложения. Проверка и обработка ошибок среди других функций API выходят из коробки, что может помочь в повышении удовлетворенности клиентов и надежности приложения.
Это подводит нас к последнему разделу статьи, где я продемонстрирую шаги для размещения приложения.
3. Пошаговая реализация
Давайте начнем с реального варианта использования для создания интеллектуального приложения для обработки данных.
USECASE
- Мы собираемся создать приложение для обработки данных в Python 3.8
- Приложение будет очищать веб-страницы в Интернете, чтобы получить список компаний, которые являются составляющими индекса, например, FTSE 100, S & P 500 или Dow Jones.
- Функциональность будет предоставлена через конечную точку GET Rest, которую мы будем размещать на локальном хосте через порт 8000. Конечная точка примет параметр запроса: index. Индексом может быть только один из следующих FTSE 100, S & P 500 или Dow Jones.
- Конечная точка также гарантирует, что значение является одним из принятых значений индекса. Все ошибки будут обработаны и корректно отправлены пользователям / посетителям API.
- Мы будем использовать FastAPI для автоматической генерации документации Swagger UI и OpenAPI.
1. Создайте виртуальную среду
Первый шаг - создать виртуальную среду и активировать ее.
python3 -m venv fintechexplained
fintechexplained\Scripts\activate
2. Установите зависимости
Следующим шагом является установка необходимых модулей.
Во-первых, установите фастапи
pip install fastapi == 0.58.1
В результате он также установит необходимые зависимости, такие как starlette == 0.13.4 и pydantic == 1.5.1.
Затем установите Uvicorn
pip install uvicorn == 0.11.5
Теперь, чтобы почистить веб-страницы, я собираюсь использовать библиотеку Pandas. Поэтому следующим шагом является установка pandas:
pip install pandas == 1.0.3
Давайте создадим структуру папок.
3. Создать структуру проекта
Для простоты давайте создадим следующую структуру папок.
Мы создали две папки: SmartApp и src в папке SmartApp. Кроме того, давайте создадим следующие три пустых Python файла:
main.py: этот файл будет содержать код, который запустит сервер uvicorn. Он также будет содержать код для обслуживания запроса и асинхронного возврата ответа. Этот файл также создаст экземпляр FastAPI(), который является основной точкой взаимодействия для создания API.models.py: этот файл будет содержать классы домена для приложения, такие как Company и Configuration.workflow_runner.py: этот файл будет содержать код, который будет очищать веб-страницы и возвращать названия компаний. Это наш код приложения для обработки данных, который мы собираемся представить через FastAPI.
Если вы хотите создать готовое к использованию решение, то я рекомендую использовать проект python cookie-cutter. Доступно много шаблонов, включая один, предназначенный для FastAPI.
Убедитесь, что виртуальная среда активирована.
4. Реализация кода Python
Следующий шаг включает в себя реализацию кода Python.
models.py
Откройте файл models.py и добавьте следующий код:
from enum import Enum
from pydantic import BaseModel
class MyException(Exception):
pass
class Company:
def __init__(self, symbol):
self.name = None
self.symbol = symbol
self.sector = None
self.industry = None
class Index(str, Enum):
FTSE100 = "FTSE 100"
SNP500 = "S&P 500"
DOWJONE = "Dow Jones"
class Configuration(BaseModel):
index: str = None
index_map = {
Index.FTSE100: 'https://en.wikipedia.org/wiki/FTSE_100_Index',
Index.SNP500: 'http://en.wikipedia.org/wiki/List_of_S%26P_500_companies'
}
def get_url(self):
return self.index_map[self.index]
- У нас будет класс с именем Company для представления каждой составляющей индекса.
- Мы также создаем класс конфигурации. Этот класс содержит имя индекса как Enum. Члены enum будут отображаться в документации API и в интерфейсе Swagger в виде раскрывающегося списка. Обратите внимание, что класс Configuration наследуется от pydantic BaseModel. Это позволит нам генерировать автоматическую документацию схемы данных. Класс конфигурации содержит словарь с именем index_map. Каждый элемент карты индекса имеет свой ключ в качестве члена перечисления индекса и значение в качестве URL-адреса веб-страницы с информацией о компаниях.
Также обратите внимание, я добавил функцию get_url(), которая будет возвращать URL для выбранного индекса.
- Мы также создали собственный класс Exception, MyException. Мы создадим для него обработчик исключений в
workflow_runner.py.
workflow_runner.py
Теперь, когда модели реализованы, давайте добавим код в workflow_runner.py. Это ядро нашего приложения для обработки данных.
Этот файл будет содержать код для выполнения следующих двух шагов:
- Для входного индекса, найдите его компании, выбрав соответствующую веб-страницу
- Вернуть названия компаний в виде списка строк
Обратите внимание, что метод run() в workflow_controller был объявлен как асинхронный, чтобы проиллюстрировать, как мы можем создать неблокирующую асинхронную операцию ввода-вывода.
from models import Company, Index
import pandas as pd
def get_snp_500_companies(url):
companies = []
table = pd.read_html(url)[0]
for index, row in table.iterrows():
company = Company(row['Symbol'])
company.name = row['Security']
company.sector = row['GICS Sector']
company.industry = row['GICS Sub Industry']
company.date_added = row['Date first added']
companies.append(company)
return companies
def get_ftse_companies(url):
companies = []
table = pd.read_html(url)[3]
for index, row in table.iterrows():
company = Company(row['EPIC'])
company.name = row['Company']
company.sector = row['FTSE Industry Classification Benchmark sector[12]']
companies.append(company)
return companies
async def run(config):
companies_map = {
Index.SNP500: get_snp_500_companies,
Index.FTSE100: get_ftse_companies
}
url = config.get_url()
func_to_get_data = companies_map.get(config.index, None)
if func_to_get_data is None:
raise KeyError(f'{input.index} is not suppported')
companies = func_to_get_data(url)
return [c.name for c in companies]
Я использую функцию pandas.read_html(url). Она возвращает все таблицы с веб-страницы. Поэтому я выбираю соответствующую таблицу на веб-странице. Поскольку каждая веб-страница имеет различную структуру, я зарегистрировал индексное имя с URL-адресом на карте компаний. Во время выполнения бегун рабочего процесса выберет соответствующий URL-адрес и создаст экземпляры класса компании, выполнив специальную функцию индекса.
Если мы выберем Dow Jone, это вызовет исключение. Это поможет мне продемонстрировать, как мы можем отображать дружественные пользователю ошибки.
Теперь мы готовы интегрировать FastAPI.
main.py
Наконец, мы собираемся добавить код в файл main.py
Сначала добавьте импорт:
import uvicorn
from fastapi import FastAPI, Path, Request
from fastapi.responses import JSONResponse
import workflow_runner
from models import MyException, Configuration, Index
Затем создайте экземпляр FastAPI() и объявите его как переменную приложения.
Мы даем ему название, описание и версию. Эта информация будет автоматически отображаться в документации.
app = FastAPI(title='SMART Data Science Application',
description='A Smart Data Science Application running on FastAPI + uvicorn',
version='0.0.1')
Предоставьте конечную точку GET для обработки запроса, который вызывает workflow_runner.run():
@app.get("/{index}")
async def get_result(index: Index = Path(..., title="The name of the Index")
):
config = Configuration(
index=index
)
try:
result = await workflow_runner.run(config)
return JSONResponse(status_code=200, content=result)
except Exception as e:
raise MyException(e)
Обратите внимание, как я ожидаю на запуск функции.
По сути, @app.get является операцией пути REST. Он обрабатывает запросы с помощью пути «/» и операции GET. Вызывает асинхронную функцию get_result(). Индекс параметра теперь является параметром запроса. @app.get является операцией пути REST. Он обрабатывает запросы с помощью пути «/» и операции GET. Вызывает асинхронную функцию get_result(). Индекс параметра теперь является параметром запроса.
Мы также можем пометить параметры запроса как необязательные, и документация будет автоматически отражена с учетом изменений. Мы объявили title значение метаданных для индекса параметра пути. Это появится в автоматической документации.
Мы также можем добавить правило валидации в Path(), например, gt = 0, le = 1000 для целых параметров.
Обратите внимание, функция get_result() объявлена как асинхронная. Это позволяет нам обслуживать запросы, не блокируя другие вызовы.
В результате эта функция позволяет приложению обслуживать запросы одновременно.
Последнее, на что следует обратить внимание, это то, что все исключения перехвачены, а затем возникает MyException.
Хотя лучшая практика заключается в том, чтобы явно перехватывать исключения и разбираться с ними по отдельности, но для демонстрационной цели, я перехватываю все исключения с помощью try / except.
Наконец, для обработки ошибок мы определяем конечную точку, которая может обрабатывать исключения типа MyException.
@app.exception_handler(MyException)
async def unicorn_exception_handler(request: Request, exc: MyException):
return JSONResponse(
status_code=418,
content={"message": f"Error occurred! Please contact the system admin."},
)
Теперь, когда мы получаем исключение типа MyException, оно будет отображать сообщение «Error occurred! Please contact the system admin». Это стало возможным благодаря app.exception_handler. FastAPI также позволяет нам создавать наши собственные обработчики исключений.
Наконец, добавьте код для запуска сервера uvicorn.
Обратите внимание, что мы используем порт 8000, а IP-адрес хоста - localhost (0.0.0.0).
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Мы закончили с реализацией.
5. Запуск серверного приложения
Давайте запустим приложение. Перейдите в папку, где находится main.py, и выполните следующую строку в терминале:
uvicorn main:app — reload
- Основным является имя питон файла, который мы хотим запустить.
- Приложение является объектом FastAPI внутри main.py
Reload позволит перезапустить сервер после изменения коды, которые позволяют быстрое развитие в процессе разработки.
Запуск команды покажет:
←[32mINFO←[0m: Uvicorn running on ←[1mhttp://127.0.0.1:8000←[0m (Press CTRL+C to quit)
←[32mINFO←[0m: Started reloader process [←[36m←[1m1180←[0m] using ←[36m←[1mstatreload←[0m
←[32mINFO←[0m: Started server process [←[36m7004←[0m]
←[32mINFO←[0m: Waiting for application startup.
←[32mINFO←[0m: Application startup complete.
- Сервер Uvicorn выставил приложение на локальный хост, порт 8000.
- Плюс он создал Swagger UI для приложения
- Наконец, он сгенерировал документацию OpenAPI для приложения.
6. Использование Сервиса
Во-первых, давайте рассмотрим автоматическую документацию API
Перейдите к
http://127.0.0.1:8000/redoc
Он отобразит пользовательский интерфейс Swagger.
При нажатии на index отображается список указателей:
При запуске он показывает названия компаний в ответе:
Давайте выберем Dow Jones и посмотрим, получим ли мы дружественную ошибку:
Это было сделано автоматически FastAPI.
Вызов приложения Smart Data Science из другого приложения
Мы можем позвонить Curl:
curl -X GET "http://localhost:8000/FTSE%20100" -H "accept: application/json"
Мы можем открыть браузер и перейти к URL-адресу для просмотра ответа:
http://localhost:8000/FTSE%20100
Вы можете скачать спецификацию openAPI, перейдя по ссылке:
http://localhost:8000/openapi.json
Мы видим, что проверка была встроена вместе с допустимыми значениями индекса, как показано ниже. Схема OpenAPI поддерживает интерфейс Swagger. Он включает в себя API и схему данных, как показано ниже:
{
"openapi": "3.0.2",
"info": {
"title": "SMART Data Science Application",
"description": "A Smart Data Science Application running on FastAPI + uvicorn",
"version": "0.0.1"
},
"paths": {
"/{index}": {
"get": {
"summary": "Get Result",
"operationId": "get_result__index__get",
"parameters": [
{
"required": true,
"schema": {
"title": "The name of the Index",
"enum": [
"FTSE 100",
"S&P 500",
"Dow Jones"
],
"type": "string"
},
"name": "index",
"in": "path"
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {
"$ref": "#/components/schemas/ValidationError"
}
}
}
},
"ValidationError": {
"title": "ValidationError",
"required": [
"loc",
"msg",
"type"
],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {
"type": "string"
}
},
"msg": {
"title": "Message",
"type": "string"
},
"type": {
"title": "Error Type",
"type": "string"
}
}
}
}
}
}
Это документация OpenAPI для API, автоматически созданного FastAPI. Мы можем отправить его вызывающим сторонам API, чтобы они могли легко интегрировать свои услуги с нашим решением.
Обратите внимание, что члены перечисления индекса появляются в документации OpenAPI вместе с заголовком параметра и правилами проверки.
