Использование микросервисной архитектуры для построения корпоративных приложений взамен традиционной монолитной — популярный тренд в веб-разработке.
Я не ставил целью настоящей статьи познакомить читателей с концепцией микросервисов. Желающим получить общее введение в тему могу порекомендовать заглянуть сюда.
Первая проблема, которую вам предстоит решить, столкнувшись на практике с задачей написать микросервис на Python — выбор подходящего фреймворка. Можно, конечно, использовать мегапопулярный Django, но, на мой взгляд, это все равно, что забивать гвоздь при помощи гидромолота. Существуют легковесные и простые в освоении решения, специально предназначенные для построения быстрых серверов API: Flask, Tornado, FastAPI и другие.
Мой выбор — Tornado. Он асинхронный, а значит при правильном подходе к разработке — более производительный. Разрабатывается с 2009 года, а значит давно переболел всеми детскими болезнями и не преподносит сюрпризов с совместимостью при выходе новых релизов. Публикаций в сети по нему накопилось огромное количество, так что, осваивая этот инструмент, вы точно не останетесь один на один со своими проблемами.
Поработав с Tornado в паре коммерческих проектов, я в целом остался доволен результатами. Однако, как бы ни было хорошо, всегда хочется чего-то большего. Например, когда я ввожу новых разработчиков в курс дела, меня слегка напрягает необходимость давать им многословные инструкции где что лежит и куда что надо прописать, чтобы создать точку входа. Нет, можно, конечно, всю бизнес-логику хранить в одном модуле, но как только приложение выходит за рамки «Hello world!», этот подход начинает слегка напрягать. Хочется так: написал хэндлер, бросил в известную папку и забыл — сервер сам его подхватит и разберется какой адрес ему назначить. Тот же самое и в отношении статического контента. Еще хочется поддержки «из коробки» датаклассов Pydantic (как у FastAPI), быстрого переключения между http и https, бенчмаркинга и прочих приятных мелочей.
Результатом моих раздумий стала небольшая библиотека CleanAPI, представляющая собой оболочку над Tornado, позволяющую повысить удобство разработки и снизить до предела (и без того невысокий) порог вхождения для новичков. Никакого принципиально нового функционала в оригинальный фреймворк я от себя не добавил — исключительно синтаксический сахар.
Устанавливаем библиотеку:
pip cleanapiСоздаем в корне проекта следующую структуру папок:
Пишем простейший хэндлер simple_handler.py:
from cleanapi import BaseHandler
url_tail = '/example.json'
class Handler(BaseHandler):
async def get(self):
self.set_status(200)
self.write({'status': 'working'})Пишем статическую страничку index.html, например, такую:
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Демонстрация CleanAPI</title>
<link rel="icon" href="/favicon.ico" type="image/x-icon" />
</head>
<body>
<h1>Демо-страница</h1>
<p>Все в порядке!</p>
</body>
</html>Перфекционисты могут еще положить в папку static_html иконку favicon.ico
И наконец, пишем запускаемый скрипт server_example.py:
from cleanapi import BaseHandler
server.start('http', 8080, '/', './handlers', './static_html')Результатом любуемся по адресам:
http://localhost:8080
http://localhost:8080/example.json
Почти столь же просто стало использование датаклассов Pydantic. Создадим в папке handlers еще один обработчик pydantic_handler.py:
from cleanapi.server import PydanticHandler
from pydantic import BaseModel, validator, NonNegativeInt
from typing import Optional, List
url_tail = '/pydantic.json'
class PydanticRequest(BaseModel):
foo: NonNegativeInt
bar: NonNegativeInt
@validator('foo', 'bar')
def _validate_user_id(cls, val: str):
if val == 666:
raise ValueError(f'Values of foo and bar should not be equal to 666')
return val
class PydanticResponse(BaseModel):
summ: Optional[NonNegativeInt]
errors: Optional[List[dict]]
# noinspection PyAbstractClass
class Handler(PydanticHandler):
request_dataclass = PydanticRequest
result_dataclass = PydanticResponse
# noinspection PyUnusedLocal
def process(self, request: request_dataclass) -> result_dataclass:
result = PydanticResponse(summ=request.foo + request.bar, errors=[])
if result.summ > 500:
raise ValueError('The sum of foo and bar is more than 1000')
return result
def if_exception(self, errors: list) -> None:
self.set_status(400)
self.write({'errors': errors})
returnРезультат наблюдаем по адресу: http://localhost:8080/pydantic.json
Для этого надо методом POST передать запрос в формате json. Браузером это сделать не получится, надо использовать утилиту типа Postman или самописный скрипт вроде:
import requests
import json
import urllib3
from funnydeco import benchmark
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
host = 'http://localhost:8080'
url_pydantic = f'{host}/pydantic.json'
headers = {'Content-type': 'application/json'}
params_pydantic = {
"foo": 8,
"bar": 4
}
@benchmark
def requester(url: str, params: dict, print_benchmark=False, benchmark_name='') -> None:
response = requests.post(url, verify=False, data=json.dumps(params), headers=headers)
print('Server response:')
print(f'Status code - {response.status_code}')
print(json.dumps(response.json(), sort_keys=False, indent=4, ensure_ascii=False))
if __name__ == '__main__':
print_bench = True
bench_name = 'Request execution'
requester(url_pydantic, params_pydantic, print_benchmark=print_bench, benchmark_name=bench_name)Поэкспериментируйте с разными значениями foo и bar, убедитесь, что логика, заложенная в обработчике, отрабатывает как надо.
Для адаптации приложения к вашей бизнес-логике вы можете создавать свои классы хэндлеров, унаследовав их от BaseHandler или PydanticHandler.
В своем текущем проекте мне удалось за счет использования такой схемы сократить изначальный объем кода API-сервера (а он был не маленький, учитывая, что я сейчас использую более 20 точек входа) более чем в три раза.
В общем, если вам понравилось, пользуйтесь на здоровье. Придумаете свои интересные фишки — делайте пул-реквесты на https://github.com/vlakir/cleanapi.
Всем добра!
